| 1 | import { ZERO_OID, diffTrees, fromUtf8, isTreeMode, readCommit, readTree, resolvePath } from "@diffinery/git-core"; |
| 2 | import type { Actor, Conclusion, Json, JobCompletion, JobSpec, JobUsage, 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 { decryptAndReseal, 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 { localOnly } from "../auth/dev.js"; |
| 26 | import { egressFor, type Egress } from "../lib/egress.js"; |
| 27 | import { endCompute, fleetLimits, mayStart, startCompute } from "../lib/quotas.js"; |
| 28 | import { blockedBy, killSwitches, type Work } from "../lib/killswitch.js"; |
| 29 | import type { Env } from "../env.js"; |
| 30 | import type { RepoDO } from "./RepoDO.js"; |
| 31 | import { completePayload } from "./event-payloads.js"; |
| 32 | import { CACHE_IDLE_MS, CACHE_LIMIT_BYTES, artifactFileName, artifactObjectKeys } from "./runtime.js"; |
| 33 | import { atLeast, type Caller } from "./types.js"; |
| 34 | import { isJobCaller } from "./token-policy.js"; |
| 35 | import { ReusableError, expandReusable, localKey, parseCallRef, type Expanded, type Source } from "./reusable.js"; |
| 36 | import { ancestors, combineResults, jobStatus, runnerLabels, type JobNode } from "./job-graph.js"; |
| 37 | import { schedulerStub } from "../lib/schedules.js"; |
| 38 | import { APPROVAL_DAYS, type Deployment, type DeploymentReview, type Reviewer } from "./deployments.js"; |
| 39 | import { referencedSecrets, RUNNER_SECRETS, secretsFor } from "./job-secrets.js"; |
| 40 | import { chooseSize, cleanUsage, containerName, describeSize, LOOKBACK, RUNNER_SIZES, sizeLabel, sizeOfContainer, type RunnerSize, type SizedRun } from "../runner/sizes.js"; |
| 41 | |
| 42 | /** How long a sandbox token works after the sandbox starts. */ |
| 43 | export const SANDBOX_TOKEN_TTL_MS = 24 * 3600_000; |
| 44 | /** A job token works this long past the job's timeout, so a runner can still report a job that hit it. */ |
| 45 | const JOB_TOKEN_MARGIN_MS = 30 * 60_000; |
| 46 | /** |
| 47 | * How long past its timeout-minutes (from when its runner fetched it) a running job may go without |
| 48 | * reporting before Diffinery ends it itself. The runner stops a job at its timeout and reports within |
| 49 | * seconds, or within CANCEL_GRACE_MS when steps run after the timeout (`if: always()`), so this only ends |
| 50 | * jobs whose runner agent froze or vanished, which would otherwise hold their container, quota slot and |
| 51 | * minutes for up to 6 hours. |
| 52 | */ |
| 53 | export const JOB_DEADLINE_GRACE_MS = 10 * 60_000; |
| 54 | /** |
| 55 | * A cancelled hosted job's runner stops the step under way and then, like GitHub's, runs the steps that |
| 56 | * run after a cancel (`if: always()`, `if: cancelled()`, post steps) for up to 5 minutes, so cleanup such |
| 57 | * as deleting what the job created still happens. Diffinery waits this long for it to report (its 5 |
| 58 | * minutes, and time for the logs and the result), then stops its container and ends the job itself. |
| 59 | */ |
| 60 | export const CANCEL_STOP_MS = 7 * 60_000; |
| 61 | /** |
| 62 | * A run finishes when its workflow (runner/workflow.ts) has had every job's completion event. An event |
| 63 | * lost while the platform updates (completeJob's sendEvent failing during a deploy) leaves the workflow |
| 64 | * waiting up to 6 hours, and a workflow instance that failed never finishes the run: either way the run |
| 65 | * stays in progress and holds its concurrency group. Once every job of a run has been finished this long, |
| 66 | * Diffinery sends the jobs' completion events again (repairStuckRuns), so the workflow carries on with |
| 67 | * the rest of the graph. |
| 68 | */ |
| 69 | export const STUCK_RUN_MS = 5 * 60_000; |
| 70 | /** |
| 71 | * A run still stuck this long after its last job finished, though its workflow got the events again, is |
| 72 | * finished by Diffinery itself (a run whose workflow no longer runs, after STUCK_RUN_MS). |
| 73 | */ |
| 74 | export const STUCK_RUN_FINISH_MS = 30 * 60_000; |
| 75 | /** Workflow instance states in which the run's workflow can still finish the run itself. */ |
| 76 | const LIVE_WORKFLOW = new Set(["queued", "running", "paused", "waiting", "waitingForPause"]); |
| 77 | /** timeout-minutes when a job row has none (GitHub's default). */ |
| 78 | const DEFAULT_TIMEOUT_MINUTES = 360; |
| 79 | /** A job token never works longer than this after the job started (timeouts are at most 360 minutes). */ |
| 80 | export const MAX_JOB_TOKEN_TTL_MS = 7 * 3600_000; |
| 81 | |
| 82 | /** |
| 83 | * When a job's token stops working, in ms since the epoch: its timeout plus a margin after the job |
| 84 | * started (or, before it starts, after the token was issued), at most MAX_JOB_TOKEN_TTL_MS. The |
| 85 | * Workflow that times jobs out normally ends the token first; this holds even if it never does. |
| 86 | */ |
| 87 | export function jobTokenExpiry(j: Pick<JobRow, "started_at" | "token_at" | "timeout_minutes">): number | null { |
| 88 | const from = Date.parse(j.started_at ?? j.token_at ?? ""); |
| 89 | if (!Number.isFinite(from)) return null; |
| 90 | const minutes = Math.max(Number(j.timeout_minutes) || 0, 0); |
| 91 | return from + Math.min(minutes * 60_000 + JOB_TOKEN_MARGIN_MS, MAX_JOB_TOKEN_TTL_MS); |
| 92 | } |
| 93 | /** Sandboxes one person or agent may have open in a repository at once: their containers are shared by the whole instance. */ |
| 94 | export const MAX_OPEN_SANDBOXES = 5; |
| 95 | |
| 96 | const WORKFLOW_DIRS = [".github/workflows", ".diffinery/workflows"]; |
| 97 | /** Characters of text a job's log keeps; the rest is dropped after a line saying so. */ |
| 98 | export const MAX_JOB_LOG = 16 * 1048576; |
| 99 | /** Every push reads and parses a commit's workflow files: at most this many, each at most this large. */ |
| 100 | export const MAX_WORKFLOW_FILES = 256; |
| 101 | /** Jobs in one run, counting each matrix leg and each job of a called workflow: keeps a run's rows and its RunWorkflow steps bounded. */ |
| 102 | export const MAX_RUN_JOBS = 1000; |
| 103 | export const MAX_WORKFLOW_BYTES = 512 * 1024; |
| 104 | /** |
| 105 | * A hosted job whose container never got a machine (the platform had none free, usually when many |
| 106 | * jobs start at once or right after a deploy) goes back in line and starts again later, this many |
| 107 | * times, before it fails. Only jobs whose runner never reported in: nothing of theirs has run. |
| 108 | */ |
| 109 | export const MAX_PLACEMENT_TRIES = 8; |
| 110 | /** Seconds before the next start of a job whose container got no machine: 30 s, doubling to 5 minutes. */ |
| 111 | export const placementDelay = (tries: number): number => Math.min(30 * 2 ** (tries - 1), 300); |
| 112 | /** |
| 113 | * A container the platform stops after it started but before its runner fetched the job (it happens |
| 114 | * to a share of the containers a burst of jobs starts): nothing ran and little was billed, so the first |
| 115 | * time the job starts again after QUICK_RETRY_SECONDS, not counted toward MAX_PLACEMENT_TRIES. Another |
| 116 | * such stop waits like any other (placementDelay). |
| 117 | */ |
| 118 | export const QUICK_RETRIES = 1; |
| 119 | export const QUICK_RETRY_SECONDS = 5; |
| 120 | /** |
| 121 | * While a deploy's container rollout updates instances, a new container can still start on the |
| 122 | * previous runner image. A hosted runner that reports another image than the deployed one |
| 123 | * (RUNNER_IMAGE) is turned away before it runs anything, and the job starts again after |
| 124 | * STALE_IMAGE_RETRY_SECONDS, up to this many times (about 10 minutes, longer than a rollout takes); |
| 125 | * after that it runs on whatever image it got rather than wait on a stuck rollout. These waits are |
| 126 | * fixed and not counted toward MAX_PLACEMENT_TRIES: a rollout ends by itself within minutes, and |
| 127 | * backing off past its end (30 s doubling) only made jobs start late. |
| 128 | */ |
| 129 | export const STALE_IMAGE_TRIES = 30; |
| 130 | export const STALE_IMAGE_RETRY_SECONDS = 20; |
| 131 | /** |
| 132 | * A deploy's container rollout can stop a hosted runner while its job runs. A job stopped before |
| 133 | * any of its steps but checkout ran is started again from the beginning on a new runner, this many |
| 134 | * times. One stopped later fails instead and says so: its steps may have had effects (a deploy, a |
| 135 | * publish) that running them again would repeat. |
| 136 | */ |
| 137 | export const MAX_RUNNER_RESTARTS = 2; |
| 138 | /** |
| 139 | * A container start that failed on the platform's side and works when tried again: no machine free, |
| 140 | * a container sidecar or Durable Object restarting under it (seen during rollouts as "Container |
| 141 | * sidecar is shutting down"), a dropped connection to it, a container that didn't come up in time, or the |
| 142 | * platform's own "internal error; reference = …" (one failed a required check outright on 2026-10-24). |
| 143 | */ |
| 144 | const RETRYABLE_START = /no container instance|is shutting down|network connection lost|durable object reset|overloaded|too many containers|aborted waiting for container to start|internal error; reference =/i; |
| 145 | /** Hosted containers one matrix starts at once: each start takes a couple of seconds. */ |
| 146 | const START_AT_ONCE = 8; |
| 147 | /** Runs one repository may have unfinished at once; past it, new ones aren't started. */ |
| 148 | export const MAX_ACTIVE_RUNS = 500; |
| 149 | /** Self-hosted runners one repository may have registered. */ |
| 150 | export const MAX_RUNNERS = 100; |
| 151 | /** How long a runner registration token works, as on GitHub. */ |
| 152 | const REGISTRATION_TTL_MS = 3600_000; |
| 153 | /** A runner that polled this recently shows as online. */ |
| 154 | const RUNNER_ONLINE_MS = 90_000; |
| 155 | |
| 156 | /** A hosted job admitted to start (Actions.admit), and what its container starts with. */ |
| 157 | type Placement = { jobId: string; container: string; token: string; egress: Egress; vcpu: number }; |
| 158 | |
| 159 | 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 }; |
| 160 | |
| 161 | export interface RunnerInfo { |
| 162 | id: number; |
| 163 | name: string; |
| 164 | os: string; |
| 165 | status: "online" | "offline"; |
| 166 | busy: boolean; |
| 167 | ephemeral: boolean; |
| 168 | labels: Array<{ id: number; name: string; type: "read-only" | "custom" }>; |
| 169 | } |
| 170 | const SUPPORTED_LABEL = /^(ubuntu-.*|linux|diffinery(-.*)?|self-hosted|x64|amd64)$/i; |
| 171 | |
| 172 | export type RunRow = { |
| 173 | id: string; |
| 174 | run_number: number; |
| 175 | workflow_path: string; |
| 176 | workflow_name: string; |
| 177 | event: string; |
| 178 | ref: string; |
| 179 | sha: string; |
| 180 | actor: string; |
| 181 | status: "queued" | "in_progress" | "completed" | "waiting"; |
| 182 | conclusion: Conclusion | "timed_out" | null; |
| 183 | created_at: string; |
| 184 | updated_at: string; |
| 185 | pr_number: number | null; |
| 186 | attempt: number; |
| 187 | event_payload: string; |
| 188 | inputs: string; |
| 189 | concurrency_group: string | null; |
| 190 | display_title: string; |
| 191 | /** Commit the run reports checks on, when it differs from `sha` (pull request runs). */ |
| 192 | head_sha: string | null; |
| 193 | /** 1 when the code the run executes came from an agent; its jobs then act as an agent. */ |
| 194 | agent_origin?: number; |
| 195 | /** Account ids that started earlier attempts of this run (JSON list), for prevent_self_review. */ |
| 196 | prior_actors?: string | null; |
| 197 | /** What a partial re-run keeps from the attempt it re-runs (JSON RerunCarry); null for every other run. */ |
| 198 | rerun_carry?: string | null; |
| 199 | /** Commits that the run's `owner/repo/path@ref` calls resolved to when it was planned (JSON, "repoId@ref" → sha). */ |
| 200 | call_refs?: string | null; |
| 201 | }; |
| 202 | |
| 203 | /** |
| 204 | * What a partial re-run (failed jobs, or one job) keeps from the attempt it re-runs, per job key: |
| 205 | * `all` of the key's job rows, or the `legs` that aren't re-run, by their matrix (canonicalJson). |
| 206 | * startJobKey copies them into the new attempt instead of running them. Keys not listed run as usual. |
| 207 | */ |
| 208 | export type RerunCarry = Record<string, { all: string[] } | { legs: Record<string, string[]> }>; |
| 209 | |
| 210 | /** Conclusions a partial re-run keeps; a job that ended any other way runs again with "re-run failed jobs". */ |
| 211 | const KEPT_CONCLUSIONS = new Set(["success", "skipped", "neutral"]); |
| 212 | |
| 213 | /** JSON with object keys sorted, so two equal matrix combinations compare equal. */ |
| 214 | export function canonicalJson(v: unknown): string { |
| 215 | if (Array.isArray(v)) return `[${v.map(canonicalJson).join(",")}]`; |
| 216 | if (v && typeof v === "object") { |
| 217 | const o = v as Record<string, unknown>; |
| 218 | return `{${Object.keys(o) |
| 219 | .sort() |
| 220 | .map((k) => `${JSON.stringify(k)}:${canonicalJson(o[k])}`) |
| 221 | .join(",")}}`; |
| 222 | } |
| 223 | return JSON.stringify(v) ?? "null"; |
| 224 | } |
| 225 | |
| 226 | export type JobRow = { |
| 227 | id: string; |
| 228 | run_id: string; |
| 229 | job_key: string; |
| 230 | name: string; |
| 231 | /** |
| 232 | * pending: held back by its concurrency group until the group's running job finishes. |
| 233 | * approval: waiting on its environment's required reviewers or wait timer; nothing has started. |
| 234 | */ |
| 235 | status: "queued" | "pending" | "approval" | "waiting" | "in_progress" | "completed"; |
| 236 | conclusion: Conclusion | "timed_out" | null; |
| 237 | matrix: string; |
| 238 | strategy: string; |
| 239 | started_at: string | null; |
| 240 | completed_at: string | null; |
| 241 | outputs: string; |
| 242 | steps: string; |
| 243 | failure: string | null; |
| 244 | token_hash: string | null; |
| 245 | /** When token_hash was issued (null for tokens issued before it was recorded, until the schema backfills it). */ |
| 246 | token_at: string | null; |
| 247 | summary: string | null; |
| 248 | container: string | null; |
| 249 | timeout_minutes: number; |
| 250 | /** runs-on labels for self-hosted jobs (JSON array); null for Diffinery-hosted containers. */ |
| 251 | labels: string | null; |
| 252 | /** `continue-on-error` as evaluated for this matrix leg (0 or 1). */ |
| 253 | continue_on_error: number; |
| 254 | /** Job-level `concurrency` group as evaluated for this leg, and its cancel-in-progress (0 or 1). */ |
| 255 | concurrency_group: string | null; |
| 256 | concurrency_cancel: number; |
| 257 | log_chars: number; |
| 258 | fleet_held: number; |
| 259 | runner_id: number | null; |
| 260 | /** `environment:` as evaluated at launch (null for jobs without one, and for jobs launched before it was stored). */ |
| 261 | environment: string | null; |
| 262 | /** When the job started waiting on its environment's protection rules, and "released" once they passed. */ |
| 263 | gate_since: string | null; |
| 264 | gate_state: string | null; |
| 265 | placement_tries: number; |
| 266 | early_stops: number; |
| 267 | /** Times a runner on the previous image was turned away (STALE_IMAGE_TRIES), and when the job starts again, in ms since the epoch. */ |
| 268 | stale_tries?: number; |
| 269 | image_retry_at?: number | null; |
| 270 | runner_info: string | null; |
| 271 | /** Set on a job a partial re-run copied from an earlier attempt: the job that actually ran. */ |
| 272 | carried_from?: string | null; |
| 273 | /** When a running hosted job was cancelled (ms since the epoch) and why, while its runner finishes (CANCEL_STOP_MS). */ |
| 274 | cancel_at?: number | null; |
| 275 | cancel_reason?: string | null; |
| 276 | restarts: number; |
| 277 | /** Hosted runner size runs-on asked for (runner/sizes.ts); null when it is chosen from the job's earlier runs. */ |
| 278 | runner_size?: string | null; |
| 279 | /** The log line naming the size the job got and why. */ |
| 280 | size_note?: string | null; |
| 281 | /** What the job used of its hosted runner (JSON JobUsage), as the runner measured it. */ |
| 282 | usage?: string | null; |
| 283 | }; |
| 284 | |
| 285 | /** Where a job ran, as its runner reported it: the runner image, the container's location and instance, and the host's processor. */ |
| 286 | export type JobRunnerInfo = { image?: string; location?: string; instance?: string; cpu?: string }; |
| 287 | |
| 288 | /** Reads the runner's `x-diffinery-runner` header (JSON), keeping only short string fields. */ |
| 289 | export function parseRunnerInfo(header: string | null | undefined): JobRunnerInfo | null { |
| 290 | if (!header || header.length > 1000) return null; |
| 291 | try { |
| 292 | const raw = JSON.parse(header) as Record<string, unknown>; |
| 293 | const out: JobRunnerInfo = {}; |
| 294 | for (const k of ["image", "location", "instance", "cpu"] as const) if (typeof raw[k] === "string" && raw[k]) out[k] = (raw[k] as string).slice(0, 200); |
| 295 | return Object.keys(out).length ? out : null; |
| 296 | } catch { |
| 297 | return null; |
| 298 | } |
| 299 | } |
| 300 | |
| 301 | /** An environment a run's jobs wait on (GitHub's pending deployment). */ |
| 302 | export type PendingDeployment = { |
| 303 | environment: { id: number; name: string }; |
| 304 | waitTimer: number; |
| 305 | waitTimerStartedAt: string; |
| 306 | reviewers: Reviewer[]; |
| 307 | preventSelfReview: boolean; |
| 308 | /** Still waiting for a required reviewer (false once approved, or when only a timer applies). */ |
| 309 | needsReview: boolean; |
| 310 | currentUserCanApprove: boolean; |
| 311 | /** The viewer is listed but started the run, and the environment prevents self-review. */ |
| 312 | selfBlocked: boolean; |
| 313 | }; |
| 314 | |
| 315 | export type RunInfo = { |
| 316 | id: string; |
| 317 | number: number; |
| 318 | attempt: number; |
| 319 | workflowPath: string; |
| 320 | workflowName: string; |
| 321 | title: string; |
| 322 | event: string; |
| 323 | ref: string; |
| 324 | sha: string; |
| 325 | /** The commit checks are reported on: the pull request head for pull_request runs. */ |
| 326 | headSha: string; |
| 327 | /** Branch the run is about: the pull request's head branch for pull request runs. */ |
| 328 | headBranch: string; |
| 329 | actor: Actor; |
| 330 | status: RunRow["status"]; |
| 331 | conclusion: RunRow["conclusion"]; |
| 332 | createdAt: string; |
| 333 | updatedAt: string; |
| 334 | prNumber: number | null; |
| 335 | }; |
| 336 | |
| 337 | export type JobInfo = { |
| 338 | id: string; |
| 339 | runId: string; |
| 340 | key: string; |
| 341 | name: string; |
| 342 | status: JobRow["status"]; |
| 343 | conclusion: JobRow["conclusion"]; |
| 344 | matrix: Record<string, Json>; |
| 345 | startedAt: string | null; |
| 346 | completedAt: string | null; |
| 347 | outputs: Record<string, string>; |
| 348 | steps: Array<StepReport & { index: number }>; |
| 349 | failure: JobCompletion["failure"] | null; |
| 350 | summary: string | null; |
| 351 | /** The runner that took the job: a self-hosted runner's name, or the hosted container's name. */ |
| 352 | runnerName: string | null; |
| 353 | runner: JobRunnerInfo | null; |
| 354 | /** For a job a partial re-run kept from an earlier attempt (it didn't run again): the job that ran. */ |
| 355 | carriedFrom: string | null; |
| 356 | /** Times a hosted job went back in line before a runner took it (no machine, or a runner on a replaced image). */ |
| 357 | placementTries: number; |
| 358 | /** Times the platform replaced the job's runner before its steps ran, and the job started again. */ |
| 359 | restarts: number; |
| 360 | /** A hosted job's runner size (runner/sizes.ts); null for self-hosted jobs and jobs not placed yet. */ |
| 361 | runnerSize: RunnerSize | null; |
| 362 | /** What the job used of its hosted runner, as the runner measured it. */ |
| 363 | usage: JobUsage | null; |
| 364 | }; |
| 365 | |
| 366 | const toRun = (r: RunRow): RunInfo => ({ |
| 367 | id: r.id, |
| 368 | number: r.run_number, |
| 369 | attempt: r.attempt, |
| 370 | workflowPath: r.workflow_path, |
| 371 | workflowName: r.workflow_name, |
| 372 | title: r.display_title, |
| 373 | event: r.event, |
| 374 | ref: r.ref, |
| 375 | sha: r.sha, |
| 376 | headSha: r.head_sha ?? r.sha, |
| 377 | 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)\//, ""), |
| 378 | actor: JSON.parse(r.actor), |
| 379 | status: r.status, |
| 380 | conclusion: r.conclusion, |
| 381 | createdAt: r.created_at, |
| 382 | updatedAt: r.updated_at, |
| 383 | prNumber: r.pr_number, |
| 384 | }); |
| 385 | |
| 386 | const toJob = (j: JobRow): JobInfo => ({ |
| 387 | id: j.id, |
| 388 | runId: j.run_id, |
| 389 | key: j.job_key, |
| 390 | name: j.name, |
| 391 | status: j.status, |
| 392 | conclusion: j.conclusion, |
| 393 | matrix: JSON.parse(j.matrix), |
| 394 | startedAt: j.started_at, |
| 395 | completedAt: j.completed_at, |
| 396 | outputs: JSON.parse(j.outputs), |
| 397 | steps: JSON.parse(j.steps), |
| 398 | failure: j.failure ? JSON.parse(j.failure) : null, |
| 399 | summary: j.summary, |
| 400 | runnerName: j.container?.replace(/^self-hosted:/, "") ?? null, |
| 401 | runner: j.runner_info ? (JSON.parse(j.runner_info) as JobRunnerInfo) : null, |
| 402 | carriedFrom: j.carried_from ?? null, |
| 403 | placementTries: j.placement_tries ?? 0, |
| 404 | restarts: j.restarts ?? 0, |
| 405 | runnerSize: j.container && !j.container.startsWith("self-hosted:") ? sizeOfContainer(j.container) : null, |
| 406 | usage: j.usage ? (JSON.parse(j.usage) as JobUsage) : null, |
| 407 | }); |
| 408 | |
| 409 | /** |
| 410 | * GitHub Actions compatible run orchestration state. The Cloudflare Workflow |
| 411 | * (`RunWorkflow`) drives a run through these methods; the runner agent inside |
| 412 | * each job's container reports back through the runner* methods. |
| 413 | */ |
| 414 | export class Actions { |
| 415 | constructor( |
| 416 | private repo: RepoDO, |
| 417 | private env: Env, |
| 418 | ) {} |
| 419 | |
| 420 | private get sql() { |
| 421 | return this.repo.sql; |
| 422 | } |
| 423 | |
| 424 | // ---------------------------------------------------------------- workflow discovery |
| 425 | |
| 426 | async loadWorkflows(sha: string): Promise<Array<{ path: string; workflow?: Workflow; errors: string[] }>> { |
| 427 | const commit = await readCommit(this.repo.store, sha); |
| 428 | const out: Array<{ path: string; workflow?: Workflow; errors: string[] }> = []; |
| 429 | for (const dir of WORKFLOW_DIRS) { |
| 430 | const entry = await resolvePath(this.repo.store, commit.tree, dir); |
| 431 | if (!entry || !isTreeMode(entry.mode)) continue; |
| 432 | for (const f of await readTree(this.repo.store, entry.oid)) { |
| 433 | if (!/\.ya?ml$/.test(f.name) || isTreeMode(f.mode)) continue; |
| 434 | if (out.length >= MAX_WORKFLOW_FILES) break; |
| 435 | const path = `${dir}/${f.name}`; |
| 436 | if ((this.repo.store.size(f.oid) ?? 0) > MAX_WORKFLOW_BYTES) { |
| 437 | out.push({ path, errors: [`: the workflow file is larger than ${MAX_WORKFLOW_BYTES / 1024} KiB`] }); |
| 438 | continue; |
| 439 | } |
| 440 | const o = await this.repo.store.read(f.oid); |
| 441 | if (!o) continue; |
| 442 | const { workflow, errors } = parseWorkflow(fromUtf8(o.content), path); |
| 443 | out.push({ path, workflow, errors: errors.filter((e) => !isWarning(e)).map((e) => `${e.path}${e.line ? `:${e.line}` : ""}: ${e.message}`) }); |
| 444 | } |
| 445 | } |
| 446 | return out; |
| 447 | } |
| 448 | |
| 449 | async listWorkflows(): Promise<Array<{ path: string; name: string; errors: string[]; triggers: string[] }>> { |
| 450 | const sha = await this.repo.resolve("HEAD"); |
| 451 | if (!sha) return []; |
| 452 | 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) : [] })); |
| 453 | } |
| 454 | |
| 455 | /** The `repository` object of event payloads: GitHub's shape, with the repository's real visibility. */ |
| 456 | async repositoryPayload(): Promise<Record<string, unknown>> { |
| 457 | const m = this.repo.meta; |
| 458 | const row = await new Directory(this.env).repoById(m.repoId); |
| 459 | if (row) return ghRepo(this.env.PUBLIC_URL, row); |
| 460 | 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` }; |
| 461 | } |
| 462 | |
| 463 | // ---------------------------------------------------------------- triggers |
| 464 | |
| 465 | async onPush(ev: { ref: string; before: string; after: string; forced: boolean; created: boolean; deleted: boolean }, caller: Caller): Promise<void> { |
| 466 | if (ev.deleted) return; |
| 467 | const workflows = await this.loadWorkflows(ev.after); |
| 468 | let changedFiles: string[] | undefined; |
| 469 | if (ev.before !== ZERO_OID && this.repo.store.has(ev.before)) { |
| 470 | const [a, b] = await Promise.all([readCommit(this.repo.store, ev.before), readCommit(this.repo.store, ev.after)]); |
| 471 | changedFiles = (await diffTrees(this.repo.store, a.tree, b.tree)).map((c) => c.path); |
| 472 | } |
| 473 | const head = await readCommit(this.repo.store, ev.after); |
| 474 | const payload = { |
| 475 | ref: ev.ref, |
| 476 | before: ev.before, |
| 477 | after: ev.after, |
| 478 | created: ev.created, |
| 479 | deleted: ev.deleted, |
| 480 | forced: ev.forced, |
| 481 | repository: await this.repositoryPayload(), |
| 482 | pusher: { name: caller.login }, |
| 483 | sender: { login: caller.login, type: caller.kind === "agent" ? "Bot" : "User" }, |
| 484 | 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 } }, |
| 485 | commits: [{ id: ev.after, message: head.message }], |
| 486 | }; |
| 487 | if (ev.ref === `refs/heads/${this.repo.meta.defaultBranch}`) await this.syncSchedules(workflows); |
| 488 | for (const w of workflows) { |
| 489 | if (!w.workflow) { |
| 490 | if (w.errors.length) await this.createFailedRun(w.path, "push", ev.ref, ev.after, caller, w.errors); |
| 491 | continue; |
| 492 | } |
| 493 | if (!matchesTrigger(w.workflow, { name: "push", ref: ev.ref, changedFiles })) continue; |
| 494 | await this.createRun(w.workflow, { event: "push", ref: ev.ref, sha: ev.after, payload, caller, title: head.message.split("\n")[0] ?? "" }); |
| 495 | } |
| 496 | } |
| 497 | |
| 498 | /** |
| 499 | * Pull request workflows, like GitHub's: `pull_request` runs the workflow from the test |
| 500 | * merge commit (`refs/pull/<n>/merge`, the head merged into the base; nothing runs while |
| 501 | * they conflict), and `pull_request_target` runs the base branch's own workflow on the |
| 502 | * base commit, so a pull request can never change what runs with the base's secrets. |
| 503 | * Both report their checks on the pull request's head commit. |
| 504 | */ |
| 505 | async onPullRequest(action: string, number: number, caller: Caller): Promise<void> { |
| 506 | const { issue } = this.repo.collab.get(number); |
| 507 | const pull = issue.pull!; |
| 508 | // Once merged, the base already holds the changes: keep the refs as they were and let path filters pass. |
| 509 | const { baseSha, mergeSha } = pull.merged ? { baseSha: this.repo.getRef(`refs/heads/${pull.base}`), mergeSha: null } : await this.repo.collab.refreshPullRefs(number); |
| 510 | const changedFiles = pull.merged ? undefined : (await this.repo.collab.pullFiles(number)).files.map((f) => f.path); |
| 511 | const m = this.repo.meta; |
| 512 | const repository = await this.repositoryPayload(); |
| 513 | const payload = { |
| 514 | action, |
| 515 | number, |
| 516 | pull_request: { |
| 517 | number, |
| 518 | id: number, |
| 519 | state: issue.state, |
| 520 | title: issue.title, |
| 521 | body: issue.body, |
| 522 | draft: pull.draft, |
| 523 | merged: pull.merged, |
| 524 | merged_at: pull.mergedAt, |
| 525 | merged_by: pull.mergedBy ? { login: pull.mergedBy } : null, |
| 526 | merge_commit_sha: pull.merged ? pull.mergeCommit : mergeSha, |
| 527 | user: { login: issue.author.login }, |
| 528 | labels: issue.labels.map((name) => ({ name })), |
| 529 | assignees: issue.assignees.map((login) => ({ login })), |
| 530 | head: { ref: pull.head, sha: pull.headSha, repo: repository }, |
| 531 | base: { ref: pull.base, sha: baseSha, repo: repository }, |
| 532 | html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/pull/${number}`, |
| 533 | url: `${this.env.PUBLIC_URL}/api/v3/repos/${m.owner}/${m.name}/pulls/${number}`, |
| 534 | }, |
| 535 | repository, |
| 536 | sender: { login: caller.login }, |
| 537 | }; |
| 538 | const common = { payload, caller, prNumber: number, headSha: pull.headSha, headRef: pull.head, baseRef: pull.base, title: issue.title }; |
| 539 | // A merged pull request's runs use its merge commit, as on GitHub. |
| 540 | const prSha = pull.merged ? pull.mergeCommit : mergeSha; |
| 541 | if (prSha) { |
| 542 | for (const w of await this.loadWorkflows(prSha)) { |
| 543 | if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request", action, baseRef: pull.base, changedFiles })) continue; |
| 544 | await this.createRun(w.workflow, { ...common, event: "pull_request", ref: `refs/pull/${number}/merge`, sha: prSha }); |
| 545 | } |
| 546 | } |
| 547 | if (baseSha) { |
| 548 | for (const w of await this.loadWorkflows(baseSha)) { |
| 549 | if (!w.workflow || !matchesTrigger(w.workflow, { name: "pull_request_target", action, baseRef: pull.base, changedFiles })) continue; |
| 550 | await this.createRun(w.workflow, { ...common, event: "pull_request_target", ref: `refs/heads/${pull.base}`, sha: baseSha }); |
| 551 | } |
| 552 | } |
| 553 | } |
| 554 | |
| 555 | /** `merge_group` workflows for a merge queue group, run from the group's commit. */ |
| 556 | async onMergeGroup(g: { number: number; base: string; ref: string; sha: string; parent: string; title: string; sender: Actor }): Promise<void> { |
| 557 | const head = await readCommit(this.repo.store, g.sha); |
| 558 | const payload = { |
| 559 | action: "checks_requested", |
| 560 | merge_group: { |
| 561 | head_sha: g.sha, |
| 562 | head_ref: g.ref, |
| 563 | base_sha: g.parent, |
| 564 | base_ref: `refs/heads/${g.base}`, |
| 565 | head_commit: { |
| 566 | id: g.sha, |
| 567 | tree_id: head.tree, |
| 568 | message: head.message, |
| 569 | timestamp: new Date(head.committer.timestamp * 1000).toISOString(), |
| 570 | author: { name: head.author.name, email: head.author.email }, |
| 571 | committer: { name: head.committer.name, email: head.committer.email }, |
| 572 | }, |
| 573 | }, |
| 574 | repository: await this.repositoryPayload(), |
| 575 | sender: { login: g.sender.login }, |
| 576 | }; |
| 577 | for (const w of await this.loadWorkflows(g.sha)) { |
| 578 | if (!w.workflow) { |
| 579 | if (w.errors.length) await this.createFailedRun(w.path, "merge_group", g.ref, g.sha, g.sender, w.errors); |
| 580 | continue; |
| 581 | } |
| 582 | if (!matchesTrigger(w.workflow, { name: "merge_group", action: "checks_requested", baseRef: g.base })) continue; |
| 583 | 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] }); |
| 584 | } |
| 585 | } |
| 586 | |
| 587 | /** Cancel unfinished runs on a ref (a merge group branch that is going away). */ |
| 588 | async cancelRunsOnRef(ref: string, reason: string): Promise<void> { |
| 589 | 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); |
| 590 | } |
| 591 | |
| 592 | async dispatch(workflowFile: string, ref: string, inputs: Record<string, unknown>, caller: Caller): Promise<RunInfo> { |
| 593 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to run workflows"); |
| 594 | await this.refuseIfBlocked("run"); |
| 595 | const sha = await this.repo.resolve(ref); |
| 596 | if (!sha) throw new HttpError(422, `unknown ref ${ref}`); |
| 597 | const all = await this.loadWorkflows(sha); |
| 598 | const w = all.find((x) => x.path === workflowFile || x.path.endsWith(`/${workflowFile}`)); |
| 599 | if (!w?.workflow) throw new HttpError(404, `workflow ${workflowFile} not found at ${ref}${w?.errors.length ? `: ${w.errors.join("; ")}` : ""}`); |
| 600 | if (!w.workflow.on.workflow_dispatch) throw new HttpError(422, "workflow does not have a workflow_dispatch trigger"); |
| 601 | const defs = w.workflow.on.workflow_dispatch.inputs ?? {}; |
| 602 | const resolved: Record<string, unknown> = {}; |
| 603 | for (const [k, d] of Object.entries(defs)) { |
| 604 | const v = inputs[k] ?? d.default; |
| 605 | if (d.required && (v === undefined || v === "")) throw new HttpError(422, `input '${k}' is required`); |
| 606 | 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(", ")}`); |
| 607 | resolved[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 608 | } |
| 609 | const fullRef = ref.startsWith("refs/") ? ref : this.repo.getRef(`refs/heads/${ref}`) ? `refs/heads/${ref}` : `refs/tags/${ref}`; |
| 610 | const payload = { inputs: resolved, ref: fullRef, repository: await this.repositoryPayload(), sender: { login: caller.login }, workflow: w.path }; |
| 611 | const m = this.repo.meta; |
| 612 | 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() }); |
| 613 | return (await this.createRun(w.workflow, { event: "workflow_dispatch", ref: fullRef, sha, payload, caller, inputs: resolved, title: w.workflow.name ?? w.path }))!; |
| 614 | } |
| 615 | |
| 616 | async schedule(workflowPath: string, cron: string): Promise<void> { |
| 617 | const sha = await this.repo.resolve("HEAD"); |
| 618 | if (!sha) return; |
| 619 | const w = (await this.loadWorkflows(sha)).find((x) => x.path === workflowPath); |
| 620 | if (!w?.workflow || !matchesTrigger(w.workflow, { name: "schedule", cron })) return; |
| 621 | const system: Caller = { id: "system", login: "diffinery-scheduler", kind: "system", role: "write" }; |
| 622 | 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 }); |
| 623 | } |
| 624 | |
| 625 | private async syncSchedules(workflows: Array<{ path: string; workflow?: Workflow }>): Promise<void> { |
| 626 | const repoId = this.repo.meta.repoId; |
| 627 | const stmts = [this.env.DB.prepare("DELETE FROM schedules WHERE repo_id = ?").bind(repoId)]; |
| 628 | 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)); |
| 629 | await this.env.DB.batch(stmts); |
| 630 | // The scheduler's alarm starts with the first schedule, without waiting for the cron trigger. |
| 631 | if (stmts.length > 1) await schedulerStub(this.env).ensure().catch((e) => console.error("scheduler alarm check failed", e)); |
| 632 | } |
| 633 | |
| 634 | // ---------------------------------------------------------------- runs |
| 635 | |
| 636 | private nextRunNumber(): number { |
| 637 | this.sql.exec("INSERT INTO counters (name, value) VALUES ('run', 1) ON CONFLICT(name) DO UPDATE SET value = value + 1"); |
| 638 | return this.sql.exec<{ value: number }>("SELECT value FROM counters WHERE name = 'run'").one().value; |
| 639 | } |
| 640 | |
| 641 | private githubContext(run: RunRow, jobKey: string, token: string): Record<string, unknown> { |
| 642 | const m = this.repo.meta; |
| 643 | const payload = JSON.parse(run.event_payload) as Record<string, unknown>; |
| 644 | const actor = JSON.parse(run.actor) as Actor; |
| 645 | const refName = run.ref.replace(/^refs\/(heads|tags)\//, "").replace(/^refs\/pull\/(\d+)\/(head|merge)$/, "$1/merge"); |
| 646 | const pr = payload.pull_request as { head?: { ref?: string }; base?: { ref?: string } } | undefined; |
| 647 | return { |
| 648 | event_name: run.event, |
| 649 | event: payload, |
| 650 | sha: run.sha, |
| 651 | ref: run.ref, |
| 652 | ref_name: refName, |
| 653 | ref_type: run.ref.startsWith("refs/tags/") ? "tag" : "branch", |
| 654 | ref_protected: false, |
| 655 | head_ref: pr?.head?.ref ?? "", |
| 656 | base_ref: pr?.base?.ref ?? "", |
| 657 | repository: `${m.owner}/${m.name}`, |
| 658 | repository_id: String(numericId(m.repoId)), |
| 659 | repository_owner: m.owner, |
| 660 | repositoryUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 661 | actor: actor.login, |
| 662 | triggering_actor: actor.login, |
| 663 | // Numeric like GitHub's (actions parse it, e.g. download-artifact's run-id); the API accepts it as a run id. |
| 664 | run_id: String(run.run_number), |
| 665 | run_number: String(run.run_number), |
| 666 | run_attempt: String(run.attempt), |
| 667 | workflow: run.workflow_name, |
| 668 | workflow_ref: `${m.owner}/${m.name}/${run.workflow_path}@${run.ref}`, |
| 669 | job: jobKey, |
| 670 | server_url: this.env.PUBLIC_URL, |
| 671 | api_url: `${this.env.PUBLIC_URL}/api/v3`, |
| 672 | graphql_url: `${this.env.PUBLIC_URL}/api/graphql`, |
| 673 | token, |
| 674 | retention_days: "90", |
| 675 | secret_source: "Actions", |
| 676 | }; |
| 677 | } |
| 678 | |
| 679 | private async createFailedRun(path: string, event: string, ref: string, sha: string, caller: Actor, errors: string[]): Promise<void> { |
| 680 | // While Actions are suspended no run is recorded, not even a failed one. |
| 681 | if (await this.blocked("run")) return; |
| 682 | const id = newId(); |
| 683 | const t = now(); |
| 684 | this.sql.exec( |
| 685 | "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', ?, ?, ?, ?)", |
| 686 | id, |
| 687 | this.nextRunNumber(), |
| 688 | path, |
| 689 | path, |
| 690 | event, |
| 691 | ref, |
| 692 | sha, |
| 693 | JSON.stringify({ id: caller.id, login: caller.login, kind: caller.kind }), |
| 694 | t, |
| 695 | t, |
| 696 | "{}", |
| 697 | `Invalid workflow file: ${errors[0] ?? path}`, |
| 698 | ); |
| 699 | const jobId = newId(); |
| 700 | this.sql.exec( |
| 701 | "INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'invalid', 'Invalid workflow file', 'completed', 'failure', ?, ?, ?)", |
| 702 | jobId, |
| 703 | id, |
| 704 | t, |
| 705 | t, |
| 706 | JSON.stringify({ message: errors.join("\n") }), |
| 707 | ); |
| 708 | } |
| 709 | |
| 710 | async createRun( |
| 711 | wf: Workflow, |
| 712 | o: { |
| 713 | event: string; |
| 714 | ref: string; |
| 715 | sha: string; |
| 716 | headSha?: string; |
| 717 | payload: Record<string, unknown>; |
| 718 | caller: Caller | { id: string; login: string; kind: Actor["kind"] }; |
| 719 | prNumber?: number; |
| 720 | inputs?: Record<string, unknown>; |
| 721 | headRef?: string; |
| 722 | baseRef?: string; |
| 723 | title: string; |
| 724 | /** Started by a run whose code came from an agent (a re-run, or a workflow_run follower). */ |
| 725 | agentOrigin?: boolean; |
| 726 | /** Pull requests whose code the run includes (a merge group's). */ |
| 727 | prNumbers?: number[]; |
| 728 | /** A new attempt of an existing run: its number, attempt, earlier actors and what it keeps (rerun). */ |
| 729 | rerun?: { runNumber: number; attempt: number; priorActors: string[]; carry: RerunCarry | null }; |
| 730 | }, |
| 731 | ): Promise<RunInfo | null> { |
| 732 | // Like GitHub, what a job does with its GITHUB_TOKEN (pushes, pull requests) starts no new |
| 733 | // runs, so workflows cannot trigger themselves in a loop; workflow_dispatch and |
| 734 | // repository_dispatch are the exceptions. |
| 735 | if (isJobCaller(o.caller) && o.event !== "workflow_dispatch" && o.event !== "repository_dispatch") return null; |
| 736 | // A site administrator suspended Actions here (lib/killswitch.ts): events start nothing, and a |
| 737 | // dispatch, which someone asked for, is refused with the reason. |
| 738 | const block = await this.blocked("run"); |
| 739 | if (block) { |
| 740 | if (o.event === "workflow_dispatch" || o.event === "repository_dispatch") throw new HttpError(403, block); |
| 741 | return null; |
| 742 | } |
| 743 | // A dispatch loop (or a flood of pushes) stops growing here. |
| 744 | const active = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE status != 'completed'").one().n; |
| 745 | if (active >= MAX_ACTIVE_RUNS) { |
| 746 | 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`); |
| 747 | console.warn(`not starting ${wf.path}: ${active} unfinished runs`); |
| 748 | return null; |
| 749 | } |
| 750 | const id = newId(); |
| 751 | const t = now(); |
| 752 | const run: RunRow = { |
| 753 | id, |
| 754 | run_number: o.rerun?.runNumber ?? this.nextRunNumber(), |
| 755 | workflow_path: wf.path, |
| 756 | workflow_name: wf.name ?? wf.path, |
| 757 | event: o.event, |
| 758 | ref: o.ref, |
| 759 | sha: o.sha, |
| 760 | actor: JSON.stringify({ id: o.caller.id, login: o.caller.login, kind: o.caller.kind }), |
| 761 | status: "queued", |
| 762 | conclusion: null, |
| 763 | created_at: t, |
| 764 | updated_at: t, |
| 765 | pr_number: o.prNumber ?? null, |
| 766 | attempt: o.rerun?.attempt ?? 1, |
| 767 | 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)), |
| 768 | inputs: JSON.stringify(o.inputs ?? {}), |
| 769 | concurrency_group: null, |
| 770 | display_title: o.title, |
| 771 | head_sha: o.headSha && o.headSha !== o.sha ? o.headSha : null, |
| 772 | agent_origin: (await this.repo.policy.agentOrigin({ |
| 773 | caller: o.caller, |
| 774 | ref: o.ref, |
| 775 | sha: o.sha, |
| 776 | prNumbers: [...(o.prNumber !== undefined ? [o.prNumber] : []), ...(o.prNumbers ?? [])], |
| 777 | inherited: o.agentOrigin, |
| 778 | })) |
| 779 | ? 1 |
| 780 | : 0, |
| 781 | prior_actors: o.rerun ? JSON.stringify(o.rerun.priorActors) : null, |
| 782 | rerun_carry: o.rerun?.carry ? JSON.stringify(o.rerun.carry) : null, |
| 783 | }; |
| 784 | const ctx: ExprContext = { github: this.githubContext(run, "", ""), inputs: o.inputs ?? {}, vars: this.variables() }; |
| 785 | try { |
| 786 | if (wf.runName) run.display_title = interpolate(wf.runName, ctx); |
| 787 | if (wf.concurrency) run.concurrency_group = interpolate(wf.concurrency.group, ctx); |
| 788 | } catch (e) { |
| 789 | console.warn("run-name/concurrency evaluation failed", e); |
| 790 | } |
| 791 | this.sql.exec( |
| 792 | "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, prior_actors, rerun_carry) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 793 | run.id, |
| 794 | run.run_number, |
| 795 | run.workflow_path, |
| 796 | run.workflow_name, |
| 797 | run.event, |
| 798 | run.ref, |
| 799 | run.sha, |
| 800 | run.actor, |
| 801 | run.status, |
| 802 | run.created_at, |
| 803 | run.updated_at, |
| 804 | run.pr_number, |
| 805 | run.attempt, |
| 806 | run.event_payload, |
| 807 | run.inputs, |
| 808 | run.concurrency_group, |
| 809 | run.display_title, |
| 810 | run.head_sha, |
| 811 | run.agent_origin ?? 0, |
| 812 | run.prior_actors ?? null, |
| 813 | run.rerun_carry ?? null, |
| 814 | ); |
| 815 | if (run.concurrency_group && wf.concurrency) { |
| 816 | const cancelInProgress = evaluateValue(wf.concurrency.cancelInProgress, ctx); |
| 817 | const others = this.sql.exec<{ id: string; ref: string }>("SELECT id, ref FROM runs WHERE concurrency_group = ? AND status != 'completed' AND id != ?", run.concurrency_group, id).toArray(); |
| 818 | // A run the newcomer may not cancel stays, and the newcomer waits for it like any run in the group. |
| 819 | if (cancelInProgress === true || cancelInProgress === "true") { |
| 820 | for (const other of others) if (this.mayCancel(run, other)) await this.cancelInternal(other.id, "cancelled by a newer run in the same concurrency group"); |
| 821 | } else await this.replaceWaiting(run); |
| 822 | } |
| 823 | await this.env.RUN_WORKFLOW.create({ id, params: { repoId: this.repo.meta.repoId, runId: id } }); |
| 824 | const m = this.repo.meta; |
| 825 | 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 }); |
| 826 | this.repo.broadcast("runs", { action: "created", runId: id }); |
| 827 | await this.onWorkflowRun(run, "requested"); |
| 828 | return toRun(run); |
| 829 | } |
| 830 | |
| 831 | /** A run by id, or by its number (github.run_id; the latest attempt). */ |
| 832 | private runRow(id: string): RunRow { |
| 833 | const r = /^\d+$/.test(id) |
| 834 | ? this.sql.exec<RunRow>("SELECT * FROM runs WHERE run_number = ? ORDER BY attempt DESC, created_at DESC LIMIT 1", Number(id)).toArray()[0] |
| 835 | : this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ?", id).toArray()[0]; |
| 836 | if (!r) throw new HttpError(404, "run not found"); |
| 837 | return r; |
| 838 | } |
| 839 | |
| 840 | private jobRow(id: string): JobRow { |
| 841 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", id).toArray()[0]; |
| 842 | if (!j) throw new HttpError(404, "job not found"); |
| 843 | return j; |
| 844 | } |
| 845 | |
| 846 | listRuns(opts: { workflow?: string; branch?: string; status?: string; event?: string; sha?: string; prNumber?: number; limit?: number; offset?: number } = {}): { runs: RunInfo[]; total: number } { |
| 847 | const where: string[] = []; |
| 848 | const args: unknown[] = []; |
| 849 | if (opts.workflow) { |
| 850 | where.push("(workflow_path = ? OR workflow_path LIKE ?)"); |
| 851 | args.push(opts.workflow, `%/${opts.workflow}`); |
| 852 | } |
| 853 | if (opts.branch) { |
| 854 | where.push("ref = ?"); |
| 855 | args.push(`refs/heads/${opts.branch}`); |
| 856 | } |
| 857 | if (opts.status) { |
| 858 | where.push(opts.status === "completed" || opts.status === "queued" || opts.status === "in_progress" ? "status = ?" : "conclusion = ?"); |
| 859 | args.push(opts.status); |
| 860 | } |
| 861 | if (opts.event) { |
| 862 | where.push("event = ?"); |
| 863 | args.push(opts.event); |
| 864 | } |
| 865 | if (opts.sha) { |
| 866 | where.push("(sha = ? OR head_sha = ?)"); |
| 867 | args.push(opts.sha, opts.sha); |
| 868 | } |
| 869 | if (opts.prNumber) { |
| 870 | where.push("pr_number = ?"); |
| 871 | args.push(opts.prNumber); |
| 872 | } |
| 873 | const w = where.length ? `WHERE ${where.join(" AND ")}` : ""; |
| 874 | const total = this.sql.exec<{ n: number }>(`SELECT count(*) AS n FROM runs ${w}`, ...args).one().n; |
| 875 | 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(); |
| 876 | return { runs: rows.map(toRun), total }; |
| 877 | } |
| 878 | |
| 879 | getRun(id: string): RunInfo & { jobs: JobInfo[] } { |
| 880 | const run = this.runRow(id); |
| 881 | return { ...toRun(run), jobs: this.jobs(run.id) }; |
| 882 | } |
| 883 | |
| 884 | jobs(runId: string): JobInfo[] { |
| 885 | if (/^\d+$/.test(runId)) runId = this.runRow(runId).id; |
| 886 | return this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", runId).toArray().map(toJob); |
| 887 | } |
| 888 | |
| 889 | getJob(id: string): JobInfo { |
| 890 | return toJob(this.jobRow(id)); |
| 891 | } |
| 892 | |
| 893 | async logs(jobId: string, afterSeq = -1): Promise<{ lines: Array<LogLine & { seq: number }>; complete: boolean }> { |
| 894 | const j = this.jobRow(jobId); |
| 895 | // A job a partial re-run kept from an earlier attempt shows the log of the job that ran. |
| 896 | if (j.carried_from) jobId = j.carried_from; |
| 897 | 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(); |
| 898 | if (rows.length === 0 && j.status === "completed") { |
| 899 | const archived = await getFirst(this.env.BUCKET, logKeys(this.repo.meta.repoId, jobId)); |
| 900 | if (archived && afterSeq < 0) return { lines: (await archived.json()) as Array<LogLine & { seq: number }>, complete: true }; |
| 901 | } |
| 902 | return { lines: rows, complete: j.status === "completed" }; |
| 903 | } |
| 904 | |
| 905 | private logKey(jobId: string) { |
| 906 | return logKeys(this.repo.meta.repoId, jobId)[0]; |
| 907 | } |
| 908 | |
| 909 | // ---------------------------------------------------------------- orchestration (called by RunWorkflow) |
| 910 | |
| 911 | /** |
| 912 | * The run's job graph: each job with the jobs it waits for (reusable workflow calls |
| 913 | * expanded). RunWorkflow starts a job as soon as its own needs have finished. |
| 914 | */ |
| 915 | async plan(runId: string): Promise<{ jobs: JobNode[]; error?: string; concurrencyBlocked: boolean }> { |
| 916 | const run = this.runRow(runId); |
| 917 | if (run.status === "completed") return { jobs: [], error: "run already completed", concurrencyBlocked: false }; |
| 918 | let exp: Expanded; |
| 919 | try { |
| 920 | exp = await this.expanded(run); |
| 921 | } catch (e) { |
| 922 | return { jobs: [], error: (e as Error).message, concurrencyBlocked: false }; |
| 923 | } |
| 924 | const { order, errors } = jobOrder(exp.workflow); |
| 925 | if (errors.length) return { jobs: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false }; |
| 926 | if (order.flat().length > MAX_RUN_JOBS) return { jobs: [], error: `A run may have at most ${MAX_RUN_JOBS} jobs, those of called workflows included; this one has ${order.flat().length}.`, concurrencyBlocked: false }; |
| 927 | if (run.concurrency_group) { |
| 928 | // Runs that waited in the group before newer runs replaced older ones: the newest replaces them now. |
| 929 | const newest = this.sql.exec<RunRow>("SELECT * FROM runs WHERE concurrency_group = ? AND status = 'queued' AND conclusion IS NULL ORDER BY rowid DESC LIMIT 1", run.concurrency_group).toArray()[0]; |
| 930 | if (newest && newest.id !== run.id) await this.replaceWaiting(newest); |
| 931 | } |
| 932 | // Cancelled while it waited for its group: none of its jobs will run, so it finishes now instead of waiting its turn. |
| 933 | if (this.runRow(runId).conclusion === "cancelled") return { jobs: [], concurrencyBlocked: false }; |
| 934 | const holders = () => (run.concurrency_group ? 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); |
| 935 | // A run holding the group may be stuck though its jobs all finished (STUCK_RUN_MS): repair it rather than wait behind it. |
| 936 | if (holders()) await this.repairStuckRuns(); |
| 937 | const blocked = holders() > 0; |
| 938 | if (!blocked) this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'queued'", now(), runId); |
| 939 | return { jobs: order.flat().map((key) => ({ key, needs: exp.workflow.jobs[key]!.needs })), concurrencyBlocked: blocked }; |
| 940 | } |
| 941 | |
| 942 | /** The run's workflow with reusable workflow calls expanded (see ./reusable.ts). */ |
| 943 | private async expanded(run: RunRow): Promise<Expanded> { |
| 944 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path); |
| 945 | if (!wf?.workflow) throw new ReusableError(`workflow file not found or invalid: ${wf?.errors.join("; ") || run.workflow_path}`); |
| 946 | // A called workflow named by branch or tag runs the commit it pointed at when the run was planned, so a job |
| 947 | // that waits for approval runs the steps that were approved, however the branch moves meanwhile. |
| 948 | const saved = this.sql.exec<{ call_refs: string | null }>("SELECT call_refs FROM runs WHERE id = ?", run.id).toArray()[0]?.call_refs; |
| 949 | const pinned: Record<string, string> = saved ? JSON.parse(saved) : {}; |
| 950 | const before = Object.keys(pinned).length; |
| 951 | try { |
| 952 | return await expandReusable(wf.workflow, { repoId: null, sha: run.sha }, (uses, from) => this.loadCalled(uses, from, pinned)); |
| 953 | } finally { |
| 954 | // Kept even when a later call fails, so the next expansion (a re-plan while the run waits) sees the same commits. |
| 955 | if (Object.keys(pinned).length !== before) this.sql.exec("UPDATE runs SET call_refs = ? WHERE id = ?", JSON.stringify(pinned), run.id); |
| 956 | } |
| 957 | } |
| 958 | |
| 959 | /** Load a called workflow: `./path` from the same commit, or `owner/repo/path@ref` from a repository on this instance. */ |
| 960 | private async loadCalled(uses: string, from: Source, pinned: Record<string, string> = {}): Promise<{ workflow: Workflow; source: Source }> { |
| 961 | const ref = parseCallRef(uses); |
| 962 | const m = this.repo.meta; |
| 963 | let source: Source; |
| 964 | let text: string | null; |
| 965 | if (ref.kind === "local" && from.repoId === null) { |
| 966 | source = from; |
| 967 | const b = await this.repo.blob(from.sha, ref.path); |
| 968 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 969 | } else { |
| 970 | let repoId: string; |
| 971 | if (ref.kind === "local") repoId = from.repoId!; |
| 972 | else if (ref.owner.toLowerCase() === m.owner.toLowerCase() && ref.repo.toLowerCase() === m.name.toLowerCase()) repoId = m.repoId; |
| 973 | else { |
| 974 | const dir = new Directory(this.env); |
| 975 | // A transferred or renamed repository's workflows are found under the old name too. |
| 976 | const target = (await dir.repo(ref.owner, ref.repo)) ?? (await dir.movedRepo(ref.owner, ref.repo)); |
| 977 | // Like GitHub: public workflows can be shared; private ones only within the same owner. |
| 978 | 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}`); |
| 979 | repoId = target.id; |
| 980 | } |
| 981 | if (repoId === m.repoId) { |
| 982 | const sha = ref.kind === "local" ? from.sha : (pinned[`${repoId}@${ref.ref}`] ??= (await this.repo.resolve(ref.ref)) ?? ""); |
| 983 | if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""}`); |
| 984 | source = { repoId: null, sha }; |
| 985 | const b = await this.repo.blob(sha, ref.path); |
| 986 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 987 | } else { |
| 988 | const stub = this.env.REPO.get(this.env.REPO.idFromName(repoId)); |
| 989 | const sha = ref.kind === "local" ? from.sha : (pinned[`${repoId}@${ref.ref}`] ??= (await stub.resolve(ref.ref)) ?? ""); |
| 990 | if (!sha) throw new Error(`unknown ref ${ref.kind === "remote" ? ref.ref : ""} in ${ref.kind === "remote" ? `${ref.owner}/${ref.repo}` : "the called repository"}`); |
| 991 | source = { repoId, sha }; |
| 992 | const b = await stub.blob(sha, ref.path); |
| 993 | text = b && !b.binary ? fromUtf8(b.content) : null; |
| 994 | } |
| 995 | } |
| 996 | if (text === null) throw new Error(`${uses} not found`); |
| 997 | const { workflow, errors } = parseWorkflow(text, ref.path); |
| 998 | const fatal = errors.filter((e) => !isWarning(e)); |
| 999 | if (!workflow || fatal.length) throw new Error(`${uses} is invalid: ${fatal.map((e) => `${e.path}: ${e.message}`).join("; ")}`); |
| 1000 | return { workflow, source }; |
| 1001 | } |
| 1002 | |
| 1003 | /** `needs` context for a job or call: keys are as written inside its own workflow. */ |
| 1004 | private async needsResults(run: RunRow, exp: Expanded, keys: string[], scope: string | null): Promise<Record<string, NeedsResult>> { |
| 1005 | const out: Record<string, NeedsResult> = {}; |
| 1006 | for (const key of keys) out[localKey(key, scope)] = await this.resultOf(run, exp, key); |
| 1007 | return out; |
| 1008 | } |
| 1009 | |
| 1010 | /** Status functions for the `if:` of a job or call: from the results of all its ancestors, like GitHub. */ |
| 1011 | private statusOf(run: RunRow, exp: Expanded, key: string): StatusFns { |
| 1012 | const results = ancestors(exp.needs, exp.needs.get(key) ?? []).map((k) => combineResults(this.rowsOf(run, exp, k))); |
| 1013 | return jobStatus(results, run.conclusion === "cancelled"); |
| 1014 | } |
| 1015 | |
| 1016 | /** The job rows of a job, or of every job inside a reusable workflow call. */ |
| 1017 | private rowsOf(run: RunRow, exp: Expanded, key: string): Array<JobRow & { continueOnError: boolean }> { |
| 1018 | const rows = exp.calls.has(key) |
| 1019 | ? this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND substr(job_key, 1, ?) = ?", run.id, key.length + 1, `${key}/`).toArray() |
| 1020 | : this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", run.id, key).toArray(); |
| 1021 | return rows.map((r) => ({ ...r, continueOnError: !!r.continue_on_error })); |
| 1022 | } |
| 1023 | |
| 1024 | /** Result and outputs of a job, or of a reusable workflow call (from its `on.workflow_call.outputs`). */ |
| 1025 | private async resultOf(run: RunRow, exp: Expanded, key: string): Promise<NeedsResult> { |
| 1026 | const call = exp.calls.get(key); |
| 1027 | const rows = this.rowsOf(run, exp, key); |
| 1028 | const result = combineResults(rows); |
| 1029 | let outputs: Record<string, string> = {}; |
| 1030 | if (!call) for (const r of rows) Object.assign(outputs, JSON.parse(r.outputs)); |
| 1031 | if (call && Object.keys(call.workflow.on.workflow_call?.outputs ?? {}).length) { |
| 1032 | const jobs: Record<string, NeedsResult> = {}; |
| 1033 | for (const k of Object.keys(call.workflow.jobs)) jobs[k] = await this.resultOf(run, exp, `${key}/${k}`); |
| 1034 | const { inputs } = await this.callContext(run, exp, key, false); |
| 1035 | const ctx: ExprContext = { github: this.githubContext(run, localKey(key, call.parent), ""), jobs, inputs, vars: this.variables() }; |
| 1036 | outputs = {}; |
| 1037 | for (const [name, d] of Object.entries(call.workflow.on.workflow_call!.outputs!)) { |
| 1038 | try { |
| 1039 | const v = evaluateValue(d.value, ctx); |
| 1040 | outputs[name] = v === undefined || v === null ? "" : typeof v === "object" ? JSON.stringify(v) : String(v); |
| 1041 | } catch { |
| 1042 | outputs[name] = ""; |
| 1043 | } |
| 1044 | } |
| 1045 | } |
| 1046 | return { result, outputs }; |
| 1047 | } |
| 1048 | |
| 1049 | /** |
| 1050 | * `inputs` and `secrets` inside a reusable workflow call (or at the top level when |
| 1051 | * `callKey` is null): inputs from the caller's `with` and the declared defaults, |
| 1052 | * secrets from `secrets: inherit` or the caller's mapping. |
| 1053 | */ |
| 1054 | private async callContext(run: RunRow, exp: Expanded, callKey: string | null, withSecrets: boolean): Promise<{ inputs: Record<string, unknown>; secrets: Record<string, string> }> { |
| 1055 | if (callKey === null) return { inputs: JSON.parse(run.inputs), secrets: withSecrets ? { ...(await this.orgSecrets()), ...(await this.repoSecrets()) } : {} }; |
| 1056 | const call = exp.calls.get(callKey)!; |
| 1057 | const parent = await this.callContext(run, exp, call.parent, withSecrets); |
| 1058 | 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() }; |
| 1059 | const inputs: Record<string, unknown> = {}; |
| 1060 | for (const [k, d] of Object.entries(call.workflow.on.workflow_call?.inputs ?? {})) { |
| 1061 | let v = call.job.with && k in call.job.with ? evaluateValue(call.job.with[k], ctx) : evaluateValue(d.default, ctx); |
| 1062 | if (v === undefined || v === null || v === "") { |
| 1063 | if (d.required) throw new ReusableError(`jobs.${callKey}: input '${k}' is required by ${call.job.uses}`); |
| 1064 | v = d.type === "boolean" ? false : d.type === "number" ? 0 : ""; |
| 1065 | } |
| 1066 | inputs[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 1067 | } |
| 1068 | const secrets: Record<string, string> = {}; |
| 1069 | if (withSecrets) { |
| 1070 | if (call.job.secrets === "inherit") Object.assign(secrets, parent.secrets); |
| 1071 | else for (const [name, expr] of Object.entries(call.job.secrets ?? {})) secrets[name.toUpperCase()] = String(evaluateValue(expr, { ...ctx, secrets: parent.secrets }) ?? ""); |
| 1072 | for (const [name, d] of Object.entries(call.workflow.on.workflow_call?.secrets ?? {})) { |
| 1073 | if (d.required && !secrets[name.toUpperCase()]) throw new ReusableError(`jobs.${callKey}: secret '${name}' is required by ${call.job.uses}`); |
| 1074 | } |
| 1075 | } |
| 1076 | return { inputs, secrets }; |
| 1077 | } |
| 1078 | |
| 1079 | /** |
| 1080 | * Start every matrix instance of a job. Returns the ids the workflow must wait |
| 1081 | * for (instances beyond max-parallel are queued and started as others finish). |
| 1082 | */ |
| 1083 | async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> { |
| 1084 | const run = this.runRow(runId); |
| 1085 | // Finished by repairStuckRuns while its workflow was stuck: nothing more starts. |
| 1086 | if (run.status === "completed") return { jobIds: [] }; |
| 1087 | // A partial re-run copies what it keeps from the earlier attempt: the whole job, or some of its matrix legs. |
| 1088 | const carry = run.rerun_carry ? (JSON.parse(run.rerun_carry) as RerunCarry)[jobKey] : undefined; |
| 1089 | if (carry && "all" in carry) { |
| 1090 | for (const id of carry.all) this.carryJob(run, id); |
| 1091 | this.repo.broadcast(`run:${runId}`, { action: "jobs" }); |
| 1092 | return { jobIds: [] }; |
| 1093 | } |
| 1094 | const kept = new Map(Object.entries(carry && "legs" in carry ? carry.legs : {}).map(([k, ids]) => [k, [...ids]])); |
| 1095 | const exp = await this.expanded(run); |
| 1096 | const job = exp.workflow.jobs[jobKey]!; |
| 1097 | const owner = exp.workflowOf.get(jobKey)!; |
| 1098 | const parent = exp.parentOf.get(jobKey) ?? null; |
| 1099 | const chain: string[] = []; |
| 1100 | for (let c = parent; c; c = exp.calls.get(c)!.parent) chain.unshift(c); |
| 1101 | const prefix = chain.map((c) => exp.calls.get(c)!.job.name ?? localKey(c, exp.calls.get(c)!.parent)).join(" / "); |
| 1102 | const label = (name: string) => (prefix ? `${prefix} / ${name}` : name); |
| 1103 | const fail = (name: string, message: string, conclusion: Conclusion = "failure") => { |
| 1104 | const id = newId(); |
| 1105 | const t = now(); |
| 1106 | 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); |
| 1107 | this.repo.broadcast(`run:${runId}`, { action: "job", jobId: id }); |
| 1108 | return { jobIds: [] as string[] }; |
| 1109 | }; |
| 1110 | const short = localKey(jobKey, parent); |
| 1111 | let needs: Record<string, NeedsResult>; |
| 1112 | let baseCtx: ExprContext; |
| 1113 | let shouldRun: boolean; |
| 1114 | try { |
| 1115 | // The callers decide first: a skipped call, or one whose ancestors failed or were skipped, skips everything inside it. |
| 1116 | for (const c of chain) { |
| 1117 | const call = exp.calls.get(c)!; |
| 1118 | const callNeeds = await this.needsResults(run, exp, call.needs, call.parent); |
| 1119 | const outer = await this.callContext(run, exp, call.parent, false); |
| 1120 | const callCtx: ExprContext = { github: this.githubContext(run, localKey(c, call.parent), ""), needs: callNeeds, inputs: outer.inputs, vars: this.variables() }; |
| 1121 | if (!evaluateIf(call.job.if, callCtx, this.statusOf(run, exp, c))) return fail(label(job.name ?? short), "", "skipped"); |
| 1122 | } |
| 1123 | // Resolve the call's inputs and secrets now, so a missing required one fails here rather than in the runner. |
| 1124 | if (parent) await this.callContext(run, exp, parent, true); |
| 1125 | needs = await this.needsResults(run, exp, exp.needs.get(jobKey) ?? [], parent); |
| 1126 | baseCtx = { github: this.githubContext(run, short, ""), needs, inputs: (await this.callContext(run, exp, parent, false)).inputs, vars: this.variables(), env: owner.env }; |
| 1127 | shouldRun = evaluateIf(job.if, baseCtx, this.statusOf(run, exp, jobKey)); |
| 1128 | } catch (e) { |
| 1129 | return fail(label(job.name ?? short), e instanceof ReusableError ? e.message : `Error evaluating 'if': ${(e as Error).message}`); |
| 1130 | } |
| 1131 | if (!shouldRun) return fail(label(job.name ?? short), "", "skipped"); |
| 1132 | |
| 1133 | let combos: Array<Record<string, unknown>>; |
| 1134 | try { |
| 1135 | combos = expandMatrix(job.strategy?.matrix, baseCtx); |
| 1136 | } catch (e) { |
| 1137 | return fail(label(job.name ?? short), `Error expanding matrix: ${(e as Error).message}`); |
| 1138 | } |
| 1139 | if (combos.length > 256) return fail(label(job.name ?? short), "A matrix may produce at most 256 jobs."); |
| 1140 | const started = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE run_id = ?", run.id).one().n; |
| 1141 | if (started + combos.length > MAX_RUN_JOBS) return fail(label(job.name ?? short), `A run may have at most ${MAX_RUN_JOBS} jobs, matrix legs included.`); |
| 1142 | const failFast = evaluateValue(job.strategy?.failFast ?? true, baseCtx) !== false; |
| 1143 | const maxParallel = Number(evaluateValue(job.strategy?.maxParallel ?? combos.length, baseCtx)) || combos.length; |
| 1144 | const ids: string[] = []; |
| 1145 | combos.forEach((matrix, i) => { |
| 1146 | // A leg the re-run keeps isn't evaluated again: it is the earlier attempt's, and counts for neither max-parallel nor minutes. |
| 1147 | const keep = kept.get(canonicalJson(matrix))?.shift(); |
| 1148 | if (keep && this.carryJob(run, keep)) return; |
| 1149 | // Like GitHub, name, runs-on, timeout-minutes, continue-on-error and concurrency are evaluated for each matrix leg. |
| 1150 | const ctx = { ...baseCtx, matrix, strategy: { "fail-fast": failFast, "job-index": i, "job-total": combos.length, "max-parallel": maxParallel } }; |
| 1151 | let name = short; |
| 1152 | if (!job.name && Object.keys(matrix).length) name = `${short} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`; |
| 1153 | let leg: { labels: string[]; timeout: number; continueOnError: boolean; group: string | null; cancel: boolean; environment: string | null }; |
| 1154 | try { |
| 1155 | if (job.name) name = interpolate(job.name, ctx); |
| 1156 | const continueOnError = evaluateValue(job.continueOnError, ctx); |
| 1157 | const cancel = job.concurrency ? evaluateValue(job.concurrency.cancelInProgress, ctx) : false; |
| 1158 | leg = { |
| 1159 | labels: runnerLabels(job.runsOn, ctx), |
| 1160 | timeout: Number(evaluateValue(job.timeoutMinutes, ctx)) || 360, |
| 1161 | continueOnError: continueOnError === true || continueOnError === "true", |
| 1162 | group: (job.concurrency ? interpolate(job.concurrency.group, ctx).trim() : "") || null, |
| 1163 | cancel: cancel === true || cancel === "true", |
| 1164 | // Evaluated once, here: its protection rules gate the launch, and runnerSpec deploys to this same name. |
| 1165 | environment: job.environment ? interpolate(job.environment.name, ctx).trim() : null, |
| 1166 | }; |
| 1167 | } catch (e) { |
| 1168 | fail(label(name), `Error evaluating the job for this matrix combination: ${(e as Error).message}`); |
| 1169 | return; |
| 1170 | } |
| 1171 | const unsupported = leg.labels.filter((l) => !SUPPORTED_LABEL.test(l)); |
| 1172 | const selfHosted = leg.labels.some((l) => l.toLowerCase() === "self-hosted"); |
| 1173 | if (!leg.labels.length || (unsupported.length && !selfHosted)) { |
| 1174 | 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.`); |
| 1175 | return; |
| 1176 | } |
| 1177 | const id = newId(); |
| 1178 | this.sql.exec( |
| 1179 | "INSERT INTO jobs (id, run_id, job_key, name, status, matrix, strategy, timeout_minutes, labels, continue_on_error, concurrency_group, concurrency_cancel, environment, runner_size) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 1180 | id, |
| 1181 | runId, |
| 1182 | jobKey, |
| 1183 | label(name), |
| 1184 | JSON.stringify(matrix), |
| 1185 | JSON.stringify(ctx.strategy), |
| 1186 | Math.min(leg.timeout, 360), |
| 1187 | selfHosted ? JSON.stringify(leg.labels.map((l) => l.toLowerCase())) : null, |
| 1188 | leg.continueOnError ? 1 : 0, |
| 1189 | leg.group, |
| 1190 | leg.cancel ? 1 : 0, |
| 1191 | leg.environment, |
| 1192 | selfHosted ? null : sizeLabel(leg.labels), |
| 1193 | ); |
| 1194 | ids.push(id); |
| 1195 | }); |
| 1196 | await this.launchAll(ids.slice(0, maxParallel)); |
| 1197 | this.repo.broadcast(`run:${runId}`, { action: "jobs" }); |
| 1198 | return { jobIds: ids }; |
| 1199 | } |
| 1200 | |
| 1201 | /** |
| 1202 | * Copies a finished job of an earlier attempt into `run` unchanged, for a partial re-run: it |
| 1203 | * doesn't run again, start a container or use minutes, and serves the earlier job's log. Its |
| 1204 | * check run (if it had one) now points at the new attempt, with the same conclusion. |
| 1205 | */ |
| 1206 | private carryJob(run: RunRow, oldId: string): boolean { |
| 1207 | const old = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", oldId).toArray()[0]; |
| 1208 | if (!old || old.status !== "completed") return false; |
| 1209 | const id = newId(); |
| 1210 | this.sql.exec( |
| 1211 | `INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, matrix, strategy, started_at, completed_at, outputs, steps, failure, summary, container, timeout_minutes, labels, continue_on_error, concurrency_group, concurrency_cancel, log_chars, environment, runner_info, carried_from) |
| 1212 | SELECT ?, ?, job_key, name, 'completed', conclusion, matrix, strategy, started_at, completed_at, outputs, steps, failure, summary, container, timeout_minutes, labels, continue_on_error, concurrency_group, concurrency_cancel, log_chars, environment, runner_info, COALESCE(carried_from, id) FROM jobs WHERE id = ?`, |
| 1213 | id, |
| 1214 | run.id, |
| 1215 | oldId, |
| 1216 | ); |
| 1217 | if (this.sql.exec("SELECT 1 FROM checks WHERE job_id = ? AND app = 'diffinery-actions'", oldId).toArray().length) { |
| 1218 | const failure = old.failure ? (JSON.parse(old.failure) as { message: string }) : null; |
| 1219 | this.repo.upsertCheck({ |
| 1220 | sha: run.head_sha ?? run.sha, |
| 1221 | name: this.checkName(run, old), |
| 1222 | status: "completed", |
| 1223 | conclusion: old.conclusion, |
| 1224 | runId: run.id, |
| 1225 | jobId: id, |
| 1226 | detailsUrl: this.jobUrl(run, id), |
| 1227 | output: failure ? { title: "Failed", summary: failure.message } : { title: old.conclusion ?? "completed" }, |
| 1228 | }); |
| 1229 | } |
| 1230 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId: id }); |
| 1231 | return true; |
| 1232 | } |
| 1233 | |
| 1234 | private checkName(run: RunRow, job: JobRow): string { |
| 1235 | return `${run.workflow_name} / ${job.name}`; |
| 1236 | } |
| 1237 | |
| 1238 | private async launch(jobId: string): Promise<void> { |
| 1239 | const placement = await this.admit(jobId); |
| 1240 | if (placement) await this.place(placement); |
| 1241 | } |
| 1242 | |
| 1243 | /** |
| 1244 | * Starts jobs (matrix legs) together. Each is admitted (kill switches, gates, quotas, the |
| 1245 | * repository's share of the fleet) one after another, so no two take the same slot; their |
| 1246 | * containers then start START_AT_ONCE at a time, since one start takes seconds. Once a leg has to |
| 1247 | * wait for a slot, so would the rest: they wait in line with it without asking again. |
| 1248 | */ |
| 1249 | private async launchAll(jobIds: string[]): Promise<void> { |
| 1250 | const starting = new Set<Promise<void>>(); |
| 1251 | let heldAs = 0; |
| 1252 | for (const id of jobIds) { |
| 1253 | if (heldAs) { |
| 1254 | this.holdForSlot(this.jobRow(id), heldAs); |
| 1255 | continue; |
| 1256 | } |
| 1257 | const placement = await this.admit(id); |
| 1258 | if (!placement) { |
| 1259 | const j = this.jobRow(id); |
| 1260 | if (j.status === "queued" && j.fleet_held > 0) heldAs = j.fleet_held; |
| 1261 | continue; |
| 1262 | } |
| 1263 | const started: Promise<void> = this.place(placement).finally(() => starting.delete(started)); |
| 1264 | starting.add(started); |
| 1265 | if (starting.size >= START_AT_ONCE) await Promise.race(starting); |
| 1266 | } |
| 1267 | await Promise.all(starting); |
| 1268 | } |
| 1269 | |
| 1270 | /** Puts a job in line for a hosted slot: 1 the repository's share of the fleet, 2 the account's hosted jobs. */ |
| 1271 | private holdForSlot(job: JobRow, held: number): void { |
| 1272 | if (job.status === "completed") return; |
| 1273 | const run = this.runRow(job.run_id); |
| 1274 | const m = this.repo.meta; |
| 1275 | this.sql.exec("UPDATE jobs SET status = 'queued', fleet_held = ? WHERE id = ?", held, job.id); |
| 1276 | this.repo.upsertCheck({ sha: run.head_sha ?? run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId: job.id, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${job.id}` }); |
| 1277 | if (held === 2) this.repo.armAlarm(null); |
| 1278 | } |
| 1279 | |
| 1280 | /** Everything before a hosted job's container starts; null when it doesn't start now (finished, parked or held). */ |
| 1281 | private async admit(jobId: string): Promise<Placement | null> { |
| 1282 | const job = this.jobRow(jobId); |
| 1283 | if (job.status === "completed") return null; |
| 1284 | // Kill switches (lib/killswitch.ts): a suspended repository or owner starts no job; a paused instance no hosted one. |
| 1285 | const block = await this.blocked(job.labels ? "self-hosted-job" : "hosted-job"); |
| 1286 | if (block) { |
| 1287 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job was not started: ${block}.` } }); |
| 1288 | return null; |
| 1289 | } |
| 1290 | if (!(await this.gate(job)) || !(await this.concurrencySlot(job))) return null; |
| 1291 | const run = this.runRow(job.run_id); |
| 1292 | const m = this.repo.meta; |
| 1293 | if (job.labels) { |
| 1294 | // Self-hosted: park the job until a registered runner with matching labels claims it. |
| 1295 | this.sql.exec("UPDATE jobs SET status = 'waiting', container = NULL WHERE id = ?", jobId); |
| 1296 | 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}` }); |
| 1297 | return null; |
| 1298 | } |
| 1299 | // Already at an account's share of the fleet in this repository alone (lib/quotas.ts fleetLimits): |
| 1300 | // the job waits for one of the repository's own to finish. Checked here first, it needs no D1 read. |
| 1301 | 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; |
| 1302 | if (busy >= fleetLimits(this.env).share) { |
| 1303 | this.holdForSlot(job, 1); |
| 1304 | return null; |
| 1305 | } |
| 1306 | // The account that owns the repository pays (lib/quotas.ts): past its monthly minutes the job |
| 1307 | // fails; past its hosted jobs at once it waits (fleet_held = 2, retried by the alarm) for one to end. |
| 1308 | const quota = await mayStart(this.env, m.ownerId, "job").catch((e: unknown) => { |
| 1309 | console.error("quota check failed", e); |
| 1310 | return { ok: true } as const; |
| 1311 | }); |
| 1312 | if (!quota.ok) { |
| 1313 | if (quota.wait) this.holdForSlot(job, 2); |
| 1314 | else await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job was not started: ${quota.message}.` } }); |
| 1315 | return null; |
| 1316 | } |
| 1317 | this.sql.exec("UPDATE jobs SET fleet_held = 0 WHERE id = ?", jobId); |
| 1318 | const token = `djt_${m.repoId}_${jobId}_${randomToken(24)}`; |
| 1319 | const { size, note } = this.runnerSize(job, run); |
| 1320 | // A restarted job gets a new container: the replaced one may still be shutting down. |
| 1321 | const container = containerName(jobId, job.restarts, size); |
| 1322 | this.sql.exec("UPDATE jobs SET status = 'waiting', token_hash = ?, token_at = ?, container = ?, size_note = ? WHERE id = ?", await sha256(token), now(), container, note, jobId); |
| 1323 | await startCompute(this.env, { id: jobId, accountId: m.ownerId, repoId: m.repoId, kind: "job" }).catch((e: unknown) => console.error("compute usage failed", e)); |
| 1324 | 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}` }); |
| 1325 | return { jobId, container, token, egress: await egressFor(this.env, m.ownerId), vcpu: RUNNER_SIZES[size].vcpu }; |
| 1326 | } |
| 1327 | |
| 1328 | /** |
| 1329 | * The hosted runner size for a job (runner/sizes.ts): what its runs-on asks for, or else what its own |
| 1330 | * earlier runs (same workflow file and job name, on hosted runners) show it needs. A size this |
| 1331 | * deployment has no container class for runs as large. |
| 1332 | */ |
| 1333 | private runnerSize(job: JobRow, run: RunRow): { size: RunnerSize; note: string } { |
| 1334 | const pinned = (job.runner_size as RunnerSize | null | undefined) ?? null; |
| 1335 | let size: RunnerSize; |
| 1336 | let why: string; |
| 1337 | if (pinned) { |
| 1338 | size = pinned; |
| 1339 | why = "asked for by runs-on"; |
| 1340 | } else { |
| 1341 | type Row = { conclusion: string | null; container: string; usage: string | null; runner_info: string | null }; |
| 1342 | const hosted = `FROM jobs j JOIN runs r ON r.id = j.run_id |
| 1343 | WHERE r.workflow_path = ? AND j.name = ? AND j.id != ? AND j.status = 'completed' AND j.labels IS NULL AND j.container IS NOT NULL AND j.carried_from IS NULL`; |
| 1344 | const sized = (rows: Row[]): SizedRun[] => |
| 1345 | rows.map((r) => { |
| 1346 | let usage: SizedRun["usage"] = null; |
| 1347 | let cpu: string | null = null; |
| 1348 | try { |
| 1349 | usage = r.usage ? JSON.parse(r.usage) : null; |
| 1350 | } catch { |
| 1351 | /* unreadable: counts as not measured */ |
| 1352 | } |
| 1353 | try { |
| 1354 | cpu = r.runner_info ? ((JSON.parse(r.runner_info) as { cpu?: unknown }).cpu as string | undefined) ?? null : null; |
| 1355 | } catch { |
| 1356 | /* unreadable: an unknown host */ |
| 1357 | } |
| 1358 | return { conclusion: r.conclusion, size: sizeOfContainer(r.container), usage, cpu: typeof cpu === "string" ? cpu : null }; |
| 1359 | }); |
| 1360 | const history = sized( |
| 1361 | this.sql.exec<Row>(`SELECT j.conclusion, j.container, j.usage, j.runner_info ${hosted} ORDER BY j.completed_at DESC LIMIT ${LOOKBACK}`, run.workflow_path, job.name, job.id).toArray(), |
| 1362 | ); |
| 1363 | // Its typical time on large, from its last measured large runs however far back they are. |
| 1364 | const large = sized( |
| 1365 | this.sql |
| 1366 | .exec<Row>( |
| 1367 | `SELECT j.conclusion, j.container, j.usage, j.runner_info ${hosted} AND j.conclusion = 'success' AND j.usage IS NOT NULL AND j.container NOT LIKE '%.%' ORDER BY j.completed_at DESC LIMIT 10`, |
| 1368 | run.workflow_path, |
| 1369 | job.name, |
| 1370 | job.id, |
| 1371 | ) |
| 1372 | .toArray(), |
| 1373 | ); |
| 1374 | ({ size, why } = chooseSize(history, { rerun: run.attempt > 1, large })); |
| 1375 | // A job that already waited for a machine doesn't wait again for a smaller one. |
| 1376 | if (size !== "large" && job.placement_tries > 0) { |
| 1377 | size = "large"; |
| 1378 | why = "no smaller runner was free when it first started"; |
| 1379 | } |
| 1380 | } |
| 1381 | if (size !== "large" && !this.env[RUNNER_SIZES[size].binding]) { |
| 1382 | size = "large"; |
| 1383 | why = "this Diffinery has no smaller runners"; |
| 1384 | } |
| 1385 | return { size, note: describeSize(size, why, !!pinned && size === pinned) }; |
| 1386 | } |
| 1387 | |
| 1388 | /** The container class of a hosted job's container (its name says which size it is). */ |
| 1389 | private runnerNamespace(container: string) { |
| 1390 | return this.env[RUNNER_SIZES[sizeOfContainer(container)].binding] ?? this.env.RUNNER; |
| 1391 | } |
| 1392 | |
| 1393 | /** Starts an admitted job's container. One that got no machine goes back in line (placementFailed). */ |
| 1394 | private async place(p: Placement): Promise<void> { |
| 1395 | try { |
| 1396 | const ns = this.runnerNamespace(p.container); |
| 1397 | const stub = ns.get(ns.idFromName(p.container)); |
| 1398 | await stub.startJob({ repoId: this.repo.meta.repoId, jobId: p.jobId, container: p.container, token: p.token, apiUrl: this.env.PUBLIC_URL, egress: p.egress, vcpu: p.vcpu }); |
| 1399 | } catch (e) { |
| 1400 | const message = (e as Error).message; |
| 1401 | if (RETRYABLE_START.test(message)) return this.placementFailed(p.jobId, message); |
| 1402 | await this.completeJob(p.jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${message}` } }); |
| 1403 | } |
| 1404 | } |
| 1405 | |
| 1406 | |
| 1407 | // ---------------------------------------------------------------- environment protection rules |
| 1408 | |
| 1409 | private jobUrl(run: RunRow, jobId: string): string { |
| 1410 | const m = this.repo.meta; |
| 1411 | return `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`; |
| 1412 | } |
| 1413 | |
| 1414 | /** |
| 1415 | * Whose run this is, for prevent_self_review, as account ids (logins can be renamed): who started |
| 1416 | * it and its earlier attempts, and for each of them that is an agent, the account that owns it. |
| 1417 | */ |
| 1418 | async selfIds(run: RunRow): Promise<Set<string>> { |
| 1419 | const actor = JSON.parse(run.actor) as Actor; |
| 1420 | const ids = new Set([actor.id, ...((run.prior_actors ? JSON.parse(run.prior_actors) : []) as string[])].filter(Boolean)); |
| 1421 | const dir = new Directory(this.env); |
| 1422 | for (const id of [...ids]) { |
| 1423 | const a = await dir.account(id).catch(() => null); |
| 1424 | if (a?.kind === "agent" && a.owner_id) ids.add(a.owner_id); |
| 1425 | } |
| 1426 | return ids; |
| 1427 | } |
| 1428 | |
| 1429 | /** |
| 1430 | * A job whose environment has required reviewers or a wait timer waits here, before a token is |
| 1431 | * minted or a runner started, in the status `approval`: no concurrency slot, no hosted slot. |
| 1432 | * Returns whether the job may go on to start. The branch policy is checked first, so a ref that |
| 1433 | * may never deploy fails at once instead of asking anyone. |
| 1434 | */ |
| 1435 | private async gate(job: JobRow): Promise<boolean> { |
| 1436 | if (!job.environment || job.gate_state === "released") return true; |
| 1437 | const rules = this.repo.deploys.protection(job.environment); |
| 1438 | if (!rules) { |
| 1439 | if (job.status === "approval") this.release(job); |
| 1440 | return true; |
| 1441 | } |
| 1442 | const run = this.runRow(job.run_id); |
| 1443 | const fail = async (message: string) => { |
| 1444 | await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message } }); |
| 1445 | return false; |
| 1446 | }; |
| 1447 | 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.`); |
| 1448 | const decision = rules.reviewers.length ? this.repo.deploys.decision(run.id, job.environment) : null; |
| 1449 | if (decision === "rejected") { |
| 1450 | const r = this.repo.deploys.reviews(run.id).find((x) => x.environment.toLowerCase() === job.environment!.toLowerCase() && x.state === "rejected")!; |
| 1451 | return fail(`The deployment to ${job.environment} was rejected by ${r.user.login}${r.comment ? `: ${r.comment}` : "."}`); |
| 1452 | } |
| 1453 | let since = job.gate_since; |
| 1454 | if (job.status !== "approval") { |
| 1455 | since = now(); |
| 1456 | this.sql.exec("UPDATE jobs SET status = 'approval', gate_since = ? WHERE id = ?", since, job.id); |
| 1457 | this.sql.exec("UPDATE runs SET status = 'waiting', updated_at = ? WHERE id = ? AND status != 'completed'", since, run.id); |
| 1458 | const actor = JSON.parse(run.actor) as Actor; |
| 1459 | 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); |
| 1460 | 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) }); |
| 1461 | // Reviewers hear about it once per run and environment, from its first job to wait. |
| 1462 | 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; |
| 1463 | if (first && rules.reviewers.length && !decision) this.emitReview("requested", run, job.environment, rules.reviewers, actor, null, ""); |
| 1464 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId: job.id }); |
| 1465 | this.repo.broadcast("runs", { action: "waiting", runId: run.id }); |
| 1466 | } |
| 1467 | const timerEnds = Date.parse(since!) + rules.waitTimer * timerMinute(this.env); |
| 1468 | if ((rules.reviewers.length && decision !== "approved") || Date.now() < timerEnds) return false; |
| 1469 | this.release({ ...job, status: "approval" }); |
| 1470 | return true; |
| 1471 | } |
| 1472 | |
| 1473 | /** The job's protection rules passed: it goes on to start, and its run stops waiting once no other job does. */ |
| 1474 | private release(job: JobRow): void { |
| 1475 | this.sql.exec("UPDATE jobs SET gate_state = 'released', status = 'queued' WHERE id = ?", job.id); |
| 1476 | this.syncWaiting(job.run_id); |
| 1477 | } |
| 1478 | |
| 1479 | private syncWaiting(runId: string): void { |
| 1480 | if (this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND status = 'approval'", runId).toArray().length) return; |
| 1481 | this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'waiting'", now(), runId); |
| 1482 | } |
| 1483 | |
| 1484 | private async wakeGate(runId: string, jobId: string): Promise<void> { |
| 1485 | try { |
| 1486 | await (await this.env.RUN_WORKFLOW.get(runId)).sendEvent({ type: `gate-${jobId}`, payload: {} }); |
| 1487 | } catch (e) { |
| 1488 | console.error("failed to wake a job waiting for approval", e); |
| 1489 | } |
| 1490 | } |
| 1491 | |
| 1492 | private emitReview(action: "requested" | "approved" | "rejected", run: RunRow, environment: string, reviewers: Reviewer[], by: Actor, state: string | null, comment: string): void { |
| 1493 | const m = this.repo.meta; |
| 1494 | const url = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`; |
| 1495 | const ids = Object.fromEntries(this.repo.deploys.environments().map((e) => [e.name.toLowerCase(), e.id])); |
| 1496 | this.repo.emit({ |
| 1497 | type: "deployment_review", |
| 1498 | action, |
| 1499 | repoId: m.repoId, |
| 1500 | owner: m.owner, |
| 1501 | repo: m.name, |
| 1502 | payload: { |
| 1503 | since: now(), |
| 1504 | environment: environment, |
| 1505 | environment_id: ids[environment.toLowerCase()] ?? null, |
| 1506 | comment, |
| 1507 | ...(state ? { state } : {}), |
| 1508 | 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 }, |
| 1509 | reviewers: reviewers.map((r) => ({ type: "User", reviewer: { login: r.login, id: r.id } })), |
| 1510 | ...(action === "requested" ? { requestor: { login: by.login } } : { approver: { login: by.login } }), |
| 1511 | }, |
| 1512 | actor: by, |
| 1513 | at: now(), |
| 1514 | }); |
| 1515 | } |
| 1516 | |
| 1517 | /** |
| 1518 | * Where a job waiting on its environment stands, for the run's Workflow: what it waits for and |
| 1519 | * until when. Re-checks the rules first, so a job whose time came starts here. |
| 1520 | */ |
| 1521 | async jobGate(jobId: string): Promise<{ status: JobRow["status"]; wait: "review" | "timer" | null; until: string | null }> { |
| 1522 | let job = this.jobRow(jobId); |
| 1523 | if (job.status === "approval") await this.launch(jobId); |
| 1524 | job = this.jobRow(jobId); |
| 1525 | if (job.status !== "approval") return { status: job.status, wait: null, until: null }; |
| 1526 | const rules = this.repo.deploys.protection(job.environment!) ?? { reviewers: [], waitTimer: 0, preventSelfReview: false }; |
| 1527 | const since = Date.parse(job.gate_since!); |
| 1528 | const deadline = since + APPROVAL_DAYS * 86400_000; |
| 1529 | if (Date.now() >= deadline) { |
| 1530 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The deployment to ${job.environment} wasn't approved within ${APPROVAL_DAYS} days.` } }); |
| 1531 | return { status: "completed", wait: null, until: null }; |
| 1532 | } |
| 1533 | const decided = this.repo.deploys.decision(job.run_id, job.environment!) === "approved"; |
| 1534 | if (rules.reviewers.length && !decided) return { status: "approval", wait: "review", until: new Date(deadline).toISOString() }; |
| 1535 | return { status: "approval", wait: "timer", until: new Date(since + rules.waitTimer * timerMinute(this.env)).toISOString() }; |
| 1536 | } |
| 1537 | |
| 1538 | /** The environments a run's jobs wait on, with who may approve them (GitHub's pending deployments). */ |
| 1539 | async pendingDeployments(runId: string, viewer: { id: string; login: string; mayReview?: boolean } | null): Promise<PendingDeployment[]> { |
| 1540 | const run = this.runRow(runId); |
| 1541 | 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(); |
| 1542 | const self = viewer ? await this.selfIds(run) : new Set<string>(); |
| 1543 | return waiting.flatMap((w) => { |
| 1544 | const env = this.repo.deploys.environments().find((e) => e.name.toLowerCase() === w.environment.toLowerCase()); |
| 1545 | if (!env) return []; |
| 1546 | const p = env.protection; |
| 1547 | const decided = this.repo.deploys.decision(run.id, env.name); |
| 1548 | // A listed reviewer who lost access to the repository no longer decides (the caller says so in mayReview). |
| 1549 | const listed = !!viewer && viewer.mayReview !== false && p.reviewers.some((r) => r.id === viewer.id); |
| 1550 | const selfBlocked = !!viewer && p.preventSelfReview && self.has(viewer.id); |
| 1551 | return [ |
| 1552 | { |
| 1553 | environment: { id: env.id, name: env.name }, |
| 1554 | waitTimer: p.waitTimer, |
| 1555 | waitTimerStartedAt: w.since, |
| 1556 | reviewers: p.reviewers, |
| 1557 | preventSelfReview: p.preventSelfReview, |
| 1558 | needsReview: p.reviewers.length > 0 && decided !== "approved", |
| 1559 | currentUserCanApprove: listed && !selfBlocked && p.reviewers.length > 0 && !decided, |
| 1560 | selfBlocked, |
| 1561 | }, |
| 1562 | ]; |
| 1563 | }); |
| 1564 | } |
| 1565 | |
| 1566 | /** |
| 1567 | * A required reviewer approves or rejects environments a run waits on. The caller checked that |
| 1568 | * `reviewer` is a person signed in or using a write token (never an agent or a job); here the |
| 1569 | * reviewer must be listed on every environment, and not whoever started the run when the |
| 1570 | * environment prevents self-review. One approval is enough. Approved jobs start once their |
| 1571 | * wait timer has also passed; rejected ones fail. |
| 1572 | */ |
| 1573 | async reviewDeployments(runId: string, input: { environmentIds: number[]; state: "approved" | "rejected"; comment: string }, reviewer: { id: string; login: string; mayReview: boolean }): Promise<{ reviews: DeploymentReview[]; deployments: Deployment[] }> { |
| 1574 | const run = this.runRow(runId); |
| 1575 | if (!reviewer.mayReview) throw new HttpError(403, "only people with access to this repository review its deployments"); |
| 1576 | if (!input.environmentIds.length) throw new HttpError(422, "environment_ids must name at least one environment"); |
| 1577 | const pending = await this.pendingDeployments(run.id, reviewer); |
| 1578 | const chosen = input.environmentIds.map((id) => { |
| 1579 | const p = pending.find((x) => x.environment.id === id); |
| 1580 | if (!p || !p.needsReview) throw new HttpError(422, `environment ${id} isn't waiting for a review in this run`); |
| 1581 | if (p.selfBlocked) throw new HttpError(403, `${p.environment.name} doesn't let whoever started a run approve it`); |
| 1582 | if (!p.currentUserCanApprove) throw new HttpError(403, `you aren't a required reviewer for ${p.environment.name}`); |
| 1583 | return p; |
| 1584 | }); |
| 1585 | const by: Actor = { id: reviewer.id, login: reviewer.login, kind: "user" }; |
| 1586 | const out: DeploymentReview[] = []; |
| 1587 | const jobIds: string[] = []; |
| 1588 | for (const p of chosen) { |
| 1589 | out.push(this.repo.deploys.addReview(run.id, p.environment.name, input.state, input.comment, reviewer)); |
| 1590 | this.emitReview(input.state, run, p.environment.name, p.reviewers, by, input.state, input.comment); |
| 1591 | 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(); |
| 1592 | for (const j of jobs) { |
| 1593 | jobIds.push(j.id); |
| 1594 | await this.launch(j.id); |
| 1595 | await this.wakeGate(run.id, j.id); |
| 1596 | } |
| 1597 | } |
| 1598 | this.repo.broadcast(`run:${run.id}`, { action: "reviewed" }); |
| 1599 | return { reviews: out, deployments: this.repo.deploys.jobDeployments(jobIds) }; |
| 1600 | } |
| 1601 | |
| 1602 | deploymentReviews(runId: string): DeploymentReview[] { |
| 1603 | return this.repo.deploys.reviews(this.runRow(runId).id); |
| 1604 | } |
| 1605 | |
| 1606 | /** |
| 1607 | * The job's container got no machine (its start failed, or it stopped before the runner reported |
| 1608 | * in). Put the job back in line and have the run's workflow start it again after a pause, or fail it |
| 1609 | * once it has waited MAX_PLACEMENT_TRIES times. |
| 1610 | */ |
| 1611 | private async placementFailed(jobId: string, why: string, o: { quick?: boolean } = {}): Promise<void> { |
| 1612 | const job = this.jobRow(jobId); |
| 1613 | if (job.status === "completed") return; |
| 1614 | if (o.quick) { |
| 1615 | this.sql.exec("UPDATE jobs SET early_stops = early_stops + 1 WHERE id = ?", jobId); |
| 1616 | return this.backInLine(job, why, job.placement_tries, QUICK_RETRY_SECONDS, "not counted: the first stop like this is retried at once"); |
| 1617 | } |
| 1618 | const tries = job.placement_tries + 1; |
| 1619 | if (tries > MAX_PLACEMENT_TRIES) { |
| 1620 | 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.` } }); |
| 1621 | return; |
| 1622 | } |
| 1623 | await this.backInLine(job, why, tries, placementDelay(tries), `wait ${tries} of ${MAX_PLACEMENT_TRIES}`); |
| 1624 | } |
| 1625 | |
| 1626 | /** Puts a job whose runner never started back in line, says why in its log, and has the run's workflow start it again in `delay` seconds. */ |
| 1627 | private async backInLine(job: JobRow, why: string, tries: number, delay: number, count: string): Promise<void> { |
| 1628 | const jobId = job.id; |
| 1629 | this.sql.exec("UPDATE jobs SET status = 'queued', container = NULL, token_hash = NULL, placement_tries = ? WHERE id = ?", tries, jobId); |
| 1630 | await this.waiting(job, why, delay, count); |
| 1631 | try { |
| 1632 | const instance = await this.env.RUN_WORKFLOW.get(job.run_id); |
| 1633 | await instance.sendEvent({ type: `job-${jobId}`, payload: { retryInSeconds: delay } }); |
| 1634 | } catch (e) { |
| 1635 | console.error("could not ask the run's workflow to start the job again", e); |
| 1636 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${why}` } }); |
| 1637 | } |
| 1638 | } |
| 1639 | |
| 1640 | /** The log line of a job back in line, so a slow start can be told apart (no machine free, a rollout's stale image); it stops using minutes meanwhile. */ |
| 1641 | private async waiting(job: JobRow, why: string, delay: number, count: string): Promise<void> { |
| 1642 | const jobId = job.id; |
| 1643 | const seq = (this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", jobId).one().n ?? -1) + 1; |
| 1644 | const line = { t: new Date().toISOString(), step: -1, text: `##[warning]No hosted runner could start the job yet: ${why}. Diffinery tries again in ${delay} s (${count}).`, seq }; |
| 1645 | this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, ?, ?)", jobId, seq, line.t, line.step, line.text); |
| 1646 | this.sql.exec("UPDATE jobs SET log_chars = coalesce(log_chars, 0) + ? WHERE id = ?", line.text.length, jobId); |
| 1647 | this.repo.broadcast(`job:${jobId}`, { action: "logs", lines: [line] }); |
| 1648 | // Not running while it waits: it holds none of the account's hosted jobs and uses no minutes. |
| 1649 | await endCompute(this.env, jobId).catch((e: unknown) => console.error("compute usage failed", e)); |
| 1650 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 1651 | } |
| 1652 | |
| 1653 | /** |
| 1654 | * A runner on the previous image was turned away during a rollout (refuseStaleRunner): the job goes |
| 1655 | * back in line and this object starts it again after STALE_IMAGE_RETRY_SECONDS (relaunchImageHeld, |
| 1656 | * from the alarm), without the run's workflow, which keeps waiting for the job to finish. |
| 1657 | */ |
| 1658 | private async waitForImage(job: JobRow, why: string): Promise<void> { |
| 1659 | const tries = (job.stale_tries ?? 0) + 1; |
| 1660 | const at = Date.now() + STALE_IMAGE_RETRY_SECONDS * 1000; |
| 1661 | this.sql.exec("UPDATE jobs SET status = 'queued', container = NULL, token_hash = NULL, stale_tries = ?, image_retry_at = ? WHERE id = ?", tries, at, job.id); |
| 1662 | await this.waiting(job, why, STALE_IMAGE_RETRY_SECONDS, `rollout wait ${tries} of ${STALE_IMAGE_TRIES}, not counted`); |
| 1663 | this.repo.alarmBy(at); |
| 1664 | } |
| 1665 | |
| 1666 | /** When the next job waiting out a rollout starts again (waitForImage), in ms since the epoch. */ |
| 1667 | nextImageRetry(): number | null { |
| 1668 | return this.sql.exec<{ at: number | null }>("SELECT min(image_retry_at) AS at FROM jobs WHERE status = 'queued' AND image_retry_at IS NOT NULL").one().at; |
| 1669 | } |
| 1670 | |
| 1671 | /** Starts again the jobs whose rollout wait is over (waitForImage). */ |
| 1672 | async relaunchImageHeld(at = Date.now()): Promise<void> { |
| 1673 | for (const { id } of this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE status = 'queued' AND image_retry_at IS NOT NULL AND image_retry_at <= ? ORDER BY rowid", at).toArray()) { |
| 1674 | this.sql.exec("UPDATE jobs SET image_retry_at = NULL WHERE id = ?", id); |
| 1675 | await this.launch(id); |
| 1676 | } |
| 1677 | } |
| 1678 | |
| 1679 | /** The run's workflow starts a job again after placementFailed put it back in line. */ |
| 1680 | async relaunchJob(jobId: string): Promise<void> { |
| 1681 | if (this.jobRow(jobId).status === "queued") await this.launch(jobId); |
| 1682 | } |
| 1683 | |
| 1684 | /** |
| 1685 | * Job-level `concurrency`, like GitHub: at most one job of a group runs at a time in |
| 1686 | * the repository. With cancel-in-progress the newcomer cancels the group's other jobs; |
| 1687 | * without it the newcomer waits as `pending` and replaces any older pending job. |
| 1688 | * Returns whether the job may start now. |
| 1689 | */ |
| 1690 | private async concurrencySlot(job: JobRow): Promise<boolean> { |
| 1691 | if (!job.concurrency_group) return true; |
| 1692 | type Other = JobRow & { run_ref: string }; |
| 1693 | const others = this.sql |
| 1694 | .exec<Other>("SELECT j.*, r.ref AS run_ref FROM jobs j JOIN runs r ON r.id = j.run_id WHERE j.concurrency_group = ? AND j.id != ? AND j.status IN ('pending', 'waiting', 'in_progress') ORDER BY j.rowid", job.concurrency_group, job.id) |
| 1695 | .toArray(); |
| 1696 | const mine = this.sql.exec<{ ref: string }>("SELECT ref FROM runs WHERE id = ?", job.run_id).toArray()[0]; |
| 1697 | // Jobs this one may replace or cancel (ACT-7); the rest it waits behind. |
| 1698 | const cancellable = (o: Other) => !mine || this.mayCancel(mine, { ref: o.run_ref }); |
| 1699 | const pending = others.filter((o) => o.status === "pending" && cancellable(o)); |
| 1700 | const running = others.filter((o) => o.status !== "pending"); |
| 1701 | const failure = { message: `Canceling since a higher priority waiting request for '${job.concurrency_group}' exists` }; |
| 1702 | if (running.length && (!job.concurrency_cancel || !running.every(cancellable))) { |
| 1703 | this.sql.exec("UPDATE jobs SET status = 'pending' WHERE id = ?", job.id); |
| 1704 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 1705 | for (const o of pending) await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure }); |
| 1706 | return false; |
| 1707 | } |
| 1708 | // Take the group before cancelling anything, so a job that starts meanwhile waits for this one. |
| 1709 | this.sql.exec("UPDATE jobs SET status = 'waiting' WHERE id = ?", job.id); |
| 1710 | for (const o of [...pending, ...running]) { |
| 1711 | await this.stopContainer(o); |
| 1712 | await this.completeJob(o.id, { conclusion: "cancelled", outputs: {}, failure }); |
| 1713 | } |
| 1714 | return true; |
| 1715 | } |
| 1716 | |
| 1717 | /** |
| 1718 | * Like GitHub, a concurrency group without cancel-in-progress has at most one run waiting: a newer |
| 1719 | * run takes the place of older ones still waiting, which are cancelled. Otherwise every push to a |
| 1720 | * deploy's branch waits, and once the running deploy ends they start in no set order, each |
| 1721 | * deploying an older commit than the last push's (tactfolio, 2026-10-16: 15 deploys queued |
| 1722 | * behind one that hung, production went back to older commits). |
| 1723 | */ |
| 1724 | private async replaceWaiting(run: RunRow): Promise<void> { |
| 1725 | const older = this.sql |
| 1726 | .exec<{ id: string; ref: string }>("SELECT id, ref FROM runs WHERE concurrency_group = ? AND status = 'queued' AND conclusion IS NULL AND id != ? AND rowid < (SELECT rowid FROM runs WHERE id = ?)", run.concurrency_group, run.id, run.id) |
| 1727 | .toArray(); |
| 1728 | for (const o of older) if (this.mayCancel(run, o)) await this.cancelInternal(o.id, `Canceling since a higher priority waiting request for '${run.concurrency_group}' exists`); |
| 1729 | } |
| 1730 | |
| 1731 | /** |
| 1732 | * Whether a run of `by` may cancel or replace `other` in a shared concurrency group. Groups are |
| 1733 | * repository-wide, as on GitHub, but a workflow pushed to some branch could otherwise name a |
| 1734 | * deploy's group and cancel it halfway: a run on a protected branch is cancelled only by runs on |
| 1735 | * that same branch (ACT-7). |
| 1736 | */ |
| 1737 | private mayCancel(by: { ref: string }, other: { ref: string }): boolean { |
| 1738 | if (by.ref === other.ref) return true; |
| 1739 | return !(other.ref.startsWith("refs/heads/") && this.repo.protectionFor(other.ref.slice(11))); |
| 1740 | } |
| 1741 | |
| 1742 | /** Whether jobs wait on the account's hosted-job quota (RepoDO retries them on its alarm). */ |
| 1743 | hasAccountHeldJobs(): boolean { |
| 1744 | return this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE fleet_held = 2 AND status = 'queued'").one().n > 0; |
| 1745 | } |
| 1746 | |
| 1747 | /** Try again to start jobs waiting on the account's hosted-job quota, oldest first, until one waits again. */ |
| 1748 | async releaseAccountHeld(): Promise<void> { |
| 1749 | for (const { id } of this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE fleet_held = 2 AND status = 'queued' ORDER BY rowid").toArray()) { |
| 1750 | await this.launch(id); |
| 1751 | if (this.jobRow(id).fleet_held === 2) return; |
| 1752 | } |
| 1753 | } |
| 1754 | |
| 1755 | async jobTimedOut(jobId: string): Promise<void> { |
| 1756 | const j = this.jobRow(jobId); |
| 1757 | if (j.status === "completed") return; |
| 1758 | if (j.status === "queued" && j.fleet_held === 2) { |
| 1759 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: "The job waited 6 hours for a hosted runner (one of this account's hosted jobs to finish, or a free runner), and was not started." } }, "timed_out"); |
| 1760 | return; |
| 1761 | } |
| 1762 | await this.stopContainer(j); |
| 1763 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job exceeded its timeout of ${j.timeout_minutes} minutes.` } }, "timed_out"); |
| 1764 | } |
| 1765 | |
| 1766 | private earlierDeadline(a: number | null, b: number | null): number | null { |
| 1767 | return a === null ? b : b === null ? a : Math.min(a, b); |
| 1768 | } |
| 1769 | |
| 1770 | /** When a running job's runner has had its timeout and JOB_DEADLINE_GRACE_MS to report, in ms since the epoch. */ |
| 1771 | private jobDeadline(j: Pick<JobRow, "started_at" | "timeout_minutes">): number | null { |
| 1772 | const from = Date.parse(j.started_at ?? ""); |
| 1773 | if (!Number.isFinite(from)) return null; |
| 1774 | const minutes = Number(j.timeout_minutes) > 0 ? Number(j.timeout_minutes) : DEFAULT_TIMEOUT_MINUTES; |
| 1775 | return from + minutes * 60_000 + JOB_DEADLINE_GRACE_MS; |
| 1776 | } |
| 1777 | |
| 1778 | /** The earliest deadline of a running job (RepoDO's alarm wakes up for it), or null. */ |
| 1779 | nextJobDeadline(): number | null { |
| 1780 | let next: number | null = null; |
| 1781 | for (const j of this.sql.exec<JobRow>("SELECT started_at, timeout_minutes, cancel_at FROM jobs WHERE status = 'in_progress' AND started_at IS NOT NULL").toArray()) { |
| 1782 | const d = this.earlierDeadline(this.jobDeadline(j), j.cancel_at ? j.cancel_at + CANCEL_STOP_MS : null); |
| 1783 | if (d !== null && (next === null || d < next)) next = d; |
| 1784 | } |
| 1785 | return next; |
| 1786 | } |
| 1787 | |
| 1788 | /** |
| 1789 | * Ends running jobs past their deadline: their runner never reported, though it stops a job at its |
| 1790 | * timeout. Its container is stopped and the job fails as timed out, freeing the slot it held. |
| 1791 | */ |
| 1792 | /** |
| 1793 | * Unfinished runs whose jobs have all finished, with when the last one did or the run was created, |
| 1794 | * whichever is later (ms since the epoch): a re-run's carried jobs keep their earlier finish times. |
| 1795 | */ |
| 1796 | private finishedJobsRuns(): { id: string; last: number }[] { |
| 1797 | return this.sql |
| 1798 | .exec<{ id: string; last: string | null }>( |
| 1799 | `SELECT r.id, max(r.created_at, coalesce((SELECT max(completed_at) FROM jobs WHERE run_id = r.id), r.created_at)) AS last FROM runs r |
| 1800 | WHERE r.status != 'completed' AND EXISTS (SELECT 1 FROM jobs WHERE run_id = r.id) |
| 1801 | AND NOT EXISTS (SELECT 1 FROM jobs WHERE run_id = r.id AND status != 'completed')`, |
| 1802 | ) |
| 1803 | .toArray() |
| 1804 | .map((r) => ({ id: r.id, last: Date.parse(r.last ?? "") || 0 })); |
| 1805 | } |
| 1806 | |
| 1807 | /** When repairStuckRuns next has something to do, in ms since the epoch. */ |
| 1808 | nextStuckRunCheck(at = Date.now()): number | null { |
| 1809 | // Not yet due; due to finish (its workflow got the events again); or due but not repaired yet (its workflow's state was unknown): a minute later. |
| 1810 | const due = this.finishedJobsRuns().map((r) => (at < r.last + STUCK_RUN_MS ? r.last + STUCK_RUN_MS : this.nudged.has(r.id) ? r.last + STUCK_RUN_FINISH_MS : at + 60_000)); |
| 1811 | return due.length ? Math.max(at, Math.min(...due)) : null; |
| 1812 | } |
| 1813 | |
| 1814 | /** |
| 1815 | * Repairs runs stuck in progress though all their jobs finished (see STUCK_RUN_MS): after STUCK_RUN_MS |
| 1816 | * their workflow gets the jobs' completion events again, and after STUCK_RUN_FINISH_MS, or after |
| 1817 | * STUCK_RUN_MS when the workflow no longer runs, Diffinery finishes the run, which frees its concurrency group. |
| 1818 | */ |
| 1819 | async repairStuckRuns(at = Date.now()): Promise<void> { |
| 1820 | for (const r of this.finishedJobsRuns()) { |
| 1821 | if (at < r.last + STUCK_RUN_MS) continue; |
| 1822 | let instance: Awaited<ReturnType<Env["RUN_WORKFLOW"]["get"]>> | null = null; |
| 1823 | let state: string | null = null; |
| 1824 | try { |
| 1825 | instance = await this.env.RUN_WORKFLOW.get(r.id); |
| 1826 | state = (await instance.status()).status; |
| 1827 | } catch (e) { |
| 1828 | console.warn("run workflow status failed", r.id, e); |
| 1829 | } |
| 1830 | // Can't tell whether its workflow still runs: look again later (nextStuckRunCheck), up to STUCK_RUN_FINISH_MS. |
| 1831 | if (state === null && at < r.last + STUCK_RUN_FINISH_MS) continue; |
| 1832 | const live = state !== null && LIVE_WORKFLOW.has(state); |
| 1833 | if (instance && live && !this.nudged.has(r.id)) { |
| 1834 | this.nudged.add(r.id); |
| 1835 | console.warn("run stuck after its jobs finished: sending their completion events again", r.id, state); |
| 1836 | for (const j of this.sql.exec<{ id: string; conclusion: string | null }>("SELECT id, conclusion FROM jobs WHERE run_id = ?", r.id).toArray()) { |
| 1837 | await instance.sendEvent({ type: `job-${j.id}`, payload: { conclusion: j.conclusion } }).catch((e: unknown) => console.warn("resend job event failed", j.id, e)); |
| 1838 | } |
| 1839 | } |
| 1840 | if (!live || at >= r.last + STUCK_RUN_FINISH_MS) await this.finishStuckRun(r.id, state ?? "unknown"); |
| 1841 | } |
| 1842 | } |
| 1843 | |
| 1844 | /** Finishes a stuck run (repairStuckRuns). Jobs its workflow never started make it fail and are named, so it doesn't pass for a success. */ |
| 1845 | private async finishStuckRun(runId: string, state: string): Promise<void> { |
| 1846 | console.warn("run stuck after its jobs finished: finishing it", runId, state); |
| 1847 | const run = this.runRow(runId); |
| 1848 | if (run.conclusion !== "cancelled") { |
| 1849 | let missing: string[] = []; |
| 1850 | try { |
| 1851 | const started = new Set(this.sql.exec<{ job_key: string }>("SELECT DISTINCT job_key FROM jobs WHERE run_id = ?", runId).toArray().map((j) => j.job_key)); |
| 1852 | const exp = await this.expanded(run); |
| 1853 | missing = jobOrder(exp.workflow).order.flat().filter((k) => !started.has(k)).map((k) => exp.workflow.jobs[k]?.name ?? k); |
| 1854 | } catch (e) { |
| 1855 | console.warn("stuck run plan failed", runId, e); |
| 1856 | } |
| 1857 | if (missing.length) { |
| 1858 | const shown = missing.slice(0, 5).join(", ") + (missing.length > 5 ? ` and ${missing.length - 5} more` : ""); |
| 1859 | await this.plannedError(runId, `The run's workflow stopped after its jobs finished, so ${missing.length === 1 ? "a job" : `${missing.length} jobs`} never started (${shown}). Re-run the workflow.`); |
| 1860 | } |
| 1861 | } |
| 1862 | await this.finishRun(runId); |
| 1863 | } |
| 1864 | |
| 1865 | /** Runs whose completion events were sent again (repairStuckRuns), so each run gets them once per object start. */ |
| 1866 | private nudged = new Set<string>(); |
| 1867 | |
| 1868 | async enforceJobDeadlines(at = Date.now()): Promise<void> { |
| 1869 | for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE status = 'in_progress' AND started_at IS NOT NULL ORDER BY rowid").toArray()) { |
| 1870 | if (j.cancel_at && j.cancel_at + CANCEL_STOP_MS <= at) { |
| 1871 | await this.stopContainer(j); |
| 1872 | const message = `${j.cancel_reason ?? "Cancelled"}. Its runner didn't report within ${CANCEL_STOP_MS / 60_000} minutes of the cancel, so Diffinery stopped it.`; |
| 1873 | await this.completeJob(j.id, { conclusion: "cancelled", outputs: {}, failure: { message } }); |
| 1874 | continue; |
| 1875 | } |
| 1876 | const d = this.jobDeadline(j); |
| 1877 | if (d === null || d > at) continue; |
| 1878 | const minutes = Number(j.timeout_minutes) > 0 ? Number(j.timeout_minutes) : DEFAULT_TIMEOUT_MINUTES; |
| 1879 | await this.stopContainer(j); |
| 1880 | await this.completeJob( |
| 1881 | j.id, |
| 1882 | { conclusion: "failure", outputs: {}, failure: { message: `The job exceeded its timeout of ${minutes} minutes, and its runner stopped responding: Diffinery ended it ${JOB_DEADLINE_GRACE_MS / 60_000} minutes later.` } }, |
| 1883 | "timed_out", |
| 1884 | ); |
| 1885 | } |
| 1886 | } |
| 1887 | |
| 1888 | async containerStopped(jobId: string, exitCode: number | null, container?: string, requested = false): Promise<void> { |
| 1889 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", jobId).toArray()[0]; |
| 1890 | if (!j || j.status === "completed") return; |
| 1891 | // A container the job no longer runs in (it was replaced, and the job restarted in another). |
| 1892 | if (container && j.container !== container) return; |
| 1893 | // Cancelled, and its container stopped before the runner reported: the job ends as cancelled. |
| 1894 | if (j.cancel_at) return this.completeJob(jobId, { conclusion: "cancelled", outputs: {}, failure: { message: j.cancel_reason ?? "Cancelled" } }); |
| 1895 | // Already back in line (placementFailed): a container given up on, which started after all, |
| 1896 | // finds its token revoked and exits. The job's next placement starts a new one. |
| 1897 | if (j.status === "queued") return; |
| 1898 | // Stopped before the runner fetched the job, so no step ran: place it again. During a |
| 1899 | // container rollout a new container can exit (or never get a machine) before it reports in. |
| 1900 | if (j.status === "waiting" && !j.labels) { |
| 1901 | if (exitCode) return this.placementFailed(jobId, `the runner exited with code ${exitCode} before it started the job`); |
| 1902 | return this.placementFailed(jobId, "the container stopped before the runner started", { quick: !requested && (j.early_stops ?? 0) < QUICK_RETRIES }); |
| 1903 | } |
| 1904 | // Stopped without reporting while a newer runner image is deployed than the one it ran on: a rollout replaced it. |
| 1905 | const ran = j.runner_info ? (JSON.parse(j.runner_info) as JobRunnerInfo).image : undefined; |
| 1906 | if (!j.labels && ran && this.env.RUNNER_IMAGE && ran !== this.env.RUNNER_IMAGE) return this.runnerReplaced(j); |
| 1907 | // Stopped without an exit code (the library reports none, or 0) and not by Diffinery (a cancel, a |
| 1908 | // timeout): the platform stopped or lost the container, for example its host. The same as a |
| 1909 | // replacement. A runner that exits by itself reports its job first, and one that crashes exits non-zero. |
| 1910 | if (!j.labels && !exitCode && !requested) return this.runnerReplaced(j, "stopped"); |
| 1911 | 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)." } }); |
| 1912 | } |
| 1913 | |
| 1914 | /** The oldest waiting job whose runs-on labels the runner all offers, handed to it with a fresh job token. */ |
| 1915 | private async assignJob(runnerName: string, runnerLabels: string[], runnerId: number | null): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 1916 | const offered = new Set(["self-hosted", ...runnerLabels.map((l) => l.toLowerCase())]); |
| 1917 | const waiting = this.sql |
| 1918 | .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") |
| 1919 | .toArray(); |
| 1920 | const job = waiting.find((j) => (JSON.parse(j.labels!) as string[]).every((l) => offered.has(l))); |
| 1921 | if (!job) return null; |
| 1922 | // A job parked before Actions were suspended here isn't handed out. |
| 1923 | const block = await this.blocked("self-hosted-job"); |
| 1924 | if (block) { |
| 1925 | await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message: `The job was not started: ${block}.` } }); |
| 1926 | return null; |
| 1927 | } |
| 1928 | const m = this.repo.meta; |
| 1929 | const token = `djt_${m.repoId}_${job.id}_${randomToken(24)}`; |
| 1930 | const container = `self-hosted:${runnerName.slice(0, 64)}`; |
| 1931 | this.sql.exec("UPDATE jobs SET token_hash = ?, token_at = ?, container = ?, runner_id = ? WHERE id = ? AND container IS NULL", await sha256(token), now(), container, runnerId, job.id); |
| 1932 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 1933 | return { jobId: job.id, token, runId: job.run_id, name: job.name }; |
| 1934 | } |
| 1935 | |
| 1936 | // ------------------------------------------------------------ registered self-hosted runners |
| 1937 | |
| 1938 | /** |
| 1939 | * GitHub's runner registration token: a repository admin's credential, good for an hour, that a |
| 1940 | * machine trades for its own runner token. So the machine never holds a person's token, and the |
| 1941 | * jobs it runs (as the same OS user) can't read one. |
| 1942 | */ |
| 1943 | async runnerRegistrationToken(caller: Caller): Promise<{ token: string; expires_at: string }> { |
| 1944 | if (caller.kind !== "user" || caller.permissions || !atLeast(caller.role, "admin")) throw new HttpError(403, "only repository admins can add self-hosted runners"); |
| 1945 | const token = `drg_${randomToken(32)}`; |
| 1946 | const expires = new Date(Date.now() + REGISTRATION_TTL_MS).toISOString(); |
| 1947 | const hash = await sha256(token); |
| 1948 | this.sql.exec("DELETE FROM runner_registrations WHERE expires_at < ?", now()); |
| 1949 | this.sql.exec("INSERT INTO runner_registrations (token_hash, created_by, expires_at) VALUES (?, ?, ?)", hash, caller.login, expires); |
| 1950 | return { token, expires_at: expires }; |
| 1951 | } |
| 1952 | |
| 1953 | /** Register a runner with a registration token; returns its own token, shown once. */ |
| 1954 | async registerRunner(registration: string, name: string, labels: string[], ephemeral: boolean): Promise<{ id: number; name: string; labels: string[]; ephemeral: boolean; token: string; registeredBy: string }> { |
| 1955 | const hash = await sha256(registration); |
| 1956 | const reg = this.sql.exec<{ created_by: string }>("SELECT created_by FROM runner_registrations WHERE token_hash = ? AND expires_at > ?", hash, now()).toArray()[0]; |
| 1957 | if (!reg) throw new HttpError(401, "invalid or expired registration token"); |
| 1958 | const clean = [...new Set(labels.map((l) => String(l).trim().toLowerCase()).filter((l) => l && l !== "self-hosted"))]; |
| 1959 | if (clean.length > 20) throw new HttpError(422, "a runner can offer at most 20 labels"); |
| 1960 | const bad = clean.find((l) => !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(l)); |
| 1961 | if (bad) throw new HttpError(422, `invalid label ${JSON.stringify(bad.slice(0, 70))}: use letters, digits, '.', '_' and '-' (up to 64)`); |
| 1962 | const runnerName = name.trim().slice(0, 64) || "runner"; |
| 1963 | const count = this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runners").one().n; |
| 1964 | if (count >= MAX_RUNNERS) throw new HttpError(429, `this repository has ${count} self-hosted runners, the most allowed; remove one first`); |
| 1965 | 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; |
| 1966 | const token = `drn_${this.repo.meta.repoId}_${id}_${randomToken(32)}`; |
| 1967 | this.sql.exec("UPDATE runners SET token_hash = ? WHERE id = ?", await sha256(token), id); |
| 1968 | return { id, name: runnerName, labels: clean, ephemeral, token, registeredBy: reg.created_by }; |
| 1969 | } |
| 1970 | |
| 1971 | /** A registered runner asks for work with its own token. An ephemeral runner's token ends with the job it takes. */ |
| 1972 | async claimJobAsRunner(token: string): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 1973 | const id = Number(token.split("_")[2]); |
| 1974 | const r = Number.isSafeInteger(id) ? this.sql.exec<RunnerRow>("SELECT * FROM runners WHERE id = ?", id).toArray()[0] : undefined; |
| 1975 | 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"); |
| 1976 | this.sql.exec("UPDATE runners SET last_seen_at = ? WHERE id = ?", now(), r.id); |
| 1977 | const job = await this.assignJob(r.name, JSON.parse(r.labels) as string[], r.id); |
| 1978 | if (job && r.ephemeral) this.sql.exec("UPDATE runners SET token_hash = NULL WHERE id = ?", r.id); |
| 1979 | return job; |
| 1980 | } |
| 1981 | |
| 1982 | runners(caller: Caller): RunnerInfo[] { |
| 1983 | if (!atLeast(caller.role, "admin") || caller.permissions) throw new HttpError(403, "only repository admins can list self-hosted runners"); |
| 1984 | 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)); |
| 1985 | return this.sql.exec<RunnerRow>("SELECT * FROM runners ORDER BY id").toArray().map((r) => ({ |
| 1986 | id: r.id, |
| 1987 | name: r.name, |
| 1988 | os: "unknown", |
| 1989 | status: r.last_seen_at && Date.now() - Date.parse(r.last_seen_at) < RUNNER_ONLINE_MS ? "online" : "offline", |
| 1990 | busy: busy.has(r.id), |
| 1991 | ephemeral: !!r.ephemeral, |
| 1992 | labels: ["self-hosted", ...(JSON.parse(r.labels) as string[])].map((name, i) => ({ id: i + 1, name, type: i === 0 ? "read-only" : "custom" })), |
| 1993 | })); |
| 1994 | } |
| 1995 | |
| 1996 | /** Remove a runner: its token stops working at once. A job it is running carries on with its own job token. */ |
| 1997 | deleteRunner(id: number, caller: Caller): void { |
| 1998 | if (caller.kind !== "user" || caller.permissions || !atLeast(caller.role, "admin")) throw new HttpError(403, "only repository admins can remove self-hosted runners"); |
| 1999 | if (!this.sql.exec("DELETE FROM runners WHERE id = ? RETURNING id", id).toArray().length) throw new HttpError(404, "runner not found"); |
| 2000 | } |
| 2001 | |
| 2002 | /** |
| 2003 | * Stops a hosted job's container, for a job Diffinery ends itself. `graceful` instead asks its runner to |
| 2004 | * cancel the job, running the steps that run after a cancel, and report (cancelInternal). |
| 2005 | */ |
| 2006 | private async stopContainer(j: JobRow, graceful = false): Promise<void> { |
| 2007 | if (!j.container || j.container.startsWith("self-hosted:")) return; |
| 2008 | try { |
| 2009 | const ns = this.runnerNamespace(j.container); |
| 2010 | await ns.get(ns.idFromName(j.container)).stopJob(graceful); |
| 2011 | } catch (e) { |
| 2012 | console.warn("stop container failed", e); |
| 2013 | } |
| 2014 | } |
| 2015 | |
| 2016 | async finishRun(runId: string): Promise<RunInfo> { |
| 2017 | const run = this.runRow(runId); |
| 2018 | // Already finished by repairStuckRuns: the late workflow changes nothing and sends no second completed event. |
| 2019 | if (run.status === "completed") return toRun(run); |
| 2020 | const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ?", runId).toArray(); |
| 2021 | let conclusion: RunRow["conclusion"] = run.conclusion === "cancelled" ? "cancelled" : "success"; |
| 2022 | if (conclusion !== "cancelled") { |
| 2023 | // A job with continue-on-error may fail without failing the run. |
| 2024 | if (jobs.some((j) => (j.conclusion === "failure" || j.conclusion === "timed_out") && !j.continue_on_error)) conclusion = "failure"; |
| 2025 | else if (jobs.some((j) => j.conclusion === "cancelled")) conclusion = "cancelled"; |
| 2026 | // As on GitHub, a run whose jobs were all skipped is skipped, not a success, so a |
| 2027 | // workflow_run follower that checks for success (a live test after a deploy) doesn't run. |
| 2028 | else if (jobs.length && jobs.every((j) => j.conclusion === "skipped")) conclusion = "skipped"; |
| 2029 | } |
| 2030 | const t = now(); |
| 2031 | this.sql.exec("UPDATE runs SET status = 'completed', conclusion = ?, updated_at = ? WHERE id = ?", conclusion, t, runId); |
| 2032 | const m = this.repo.meta; |
| 2033 | 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 }); |
| 2034 | this.repo.broadcast(`run:${runId}`, { action: "completed", conclusion }); |
| 2035 | this.repo.broadcast("runs", { action: "completed", runId }); |
| 2036 | await this.onWorkflowRun(this.runRow(runId), "completed"); |
| 2037 | return toRun(this.runRow(runId)); |
| 2038 | } |
| 2039 | |
| 2040 | /** |
| 2041 | * Starts the default branch's `on: workflow_run` workflows that follow `run`. Like |
| 2042 | * GitHub, they run on the default branch, must name the workflows they follow, and |
| 2043 | * chains stop after three levels (A -> B -> C -> D; D's runs trigger nothing). |
| 2044 | */ |
| 2045 | private async onWorkflowRun(run: RunRow, action: "requested" | "completed"): Promise<void> { |
| 2046 | if (this.chainDepth(run) >= 3) return; |
| 2047 | try { |
| 2048 | const sha = this.repo.getRef(this.repo.headRef); |
| 2049 | if (!sha) return; |
| 2050 | const followers = (await this.loadWorkflows(sha)).filter((w) => w.workflow?.on.workflow_run?.workflows?.length); |
| 2051 | if (!followers.length) return; |
| 2052 | const m = this.repo.meta; |
| 2053 | const payload = JSON.parse(run.event_payload) as { pull_request?: { head?: { ref?: string } } }; |
| 2054 | const branch = payload.pull_request?.head?.ref ?? run.ref.replace(/^refs\/(heads|tags)\//, ""); |
| 2055 | const actor = JSON.parse(run.actor) as Actor; |
| 2056 | const workflowRun = { |
| 2057 | id: run.run_number, |
| 2058 | name: run.workflow_name, |
| 2059 | display_title: run.display_title, |
| 2060 | path: run.workflow_path, |
| 2061 | run_number: run.run_number, |
| 2062 | run_attempt: run.attempt, |
| 2063 | event: run.event, |
| 2064 | status: run.status, |
| 2065 | conclusion: run.conclusion, |
| 2066 | head_branch: branch, |
| 2067 | head_sha: run.head_sha ?? run.sha, |
| 2068 | actor: { login: actor.login }, |
| 2069 | triggering_actor: { login: actor.login }, |
| 2070 | pull_requests: run.pr_number ? [{ number: run.pr_number }] : [], |
| 2071 | created_at: run.created_at, |
| 2072 | updated_at: run.updated_at, |
| 2073 | html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}`, |
| 2074 | }; |
| 2075 | for (const w of followers) { |
| 2076 | if (!matchesTrigger(w.workflow!, { name: "workflow_run", action, workflow: run.workflow_name, branch })) continue; |
| 2077 | await this.createRun(w.workflow!, { |
| 2078 | event: "workflow_run", |
| 2079 | ref: this.repo.headRef, |
| 2080 | sha, |
| 2081 | payload: { action, workflow_run: workflowRun, workflow: { name: run.workflow_name, path: run.workflow_path }, repository: await this.repositoryPayload(), sender: { login: actor.login } }, |
| 2082 | caller: actor, |
| 2083 | title: w.workflow!.name ?? w.path, |
| 2084 | // A follower acts on what the run it follows produced (artifacts, outputs), so it stays an agent's. |
| 2085 | agentOrigin: !!run.agent_origin, |
| 2086 | }); |
| 2087 | } |
| 2088 | } catch (e) { |
| 2089 | console.warn("workflow_run trigger failed", e); |
| 2090 | } |
| 2091 | } |
| 2092 | |
| 2093 | /** How many workflow_run links led to `run` (0 for a run started by any other event). */ |
| 2094 | private chainDepth(run: RunRow): number { |
| 2095 | let depth = 0; |
| 2096 | for (let r: RunRow | undefined = run; r?.event === "workflow_run" && depth < 3; depth++) { |
| 2097 | const parent: string | number | undefined = (JSON.parse(r.event_payload) as { workflow_run?: { id?: string | number } }).workflow_run?.id; |
| 2098 | 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; |
| 2099 | } |
| 2100 | return depth; |
| 2101 | } |
| 2102 | |
| 2103 | async plannedError(runId: string, message: string): Promise<void> { |
| 2104 | const t = now(); |
| 2105 | 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 })); |
| 2106 | } |
| 2107 | |
| 2108 | // ---------------------------------------------------------------- cancellation and re-runs |
| 2109 | |
| 2110 | /** |
| 2111 | * Cancels a run's unfinished jobs. With `graceful` (someone cancelled the run), a running hosted job's |
| 2112 | * runner gets to run the steps that run after a cancel first, as on GitHub (CANCEL_STOP_MS); the job |
| 2113 | * ends when it reports. Otherwise, or when it was already cancelled that way, every job ends at once. |
| 2114 | */ |
| 2115 | private async cancelInternal(runId: string, reason: string, graceful = false): Promise<void> { |
| 2116 | const run = this.runRow(runId); |
| 2117 | if (run.status === "completed") return; |
| 2118 | this.sql.exec("UPDATE runs SET conclusion = 'cancelled', updated_at = ? WHERE id = ?", now(), run.id); |
| 2119 | // One cancel, one deadline: the jobs it lets run their after-cancel steps all end together. |
| 2120 | const at = Date.now(); |
| 2121 | for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status != 'completed'", run.id).toArray()) { |
| 2122 | if (graceful && !j.cancel_at && j.status === "in_progress" && !j.labels && j.container && !j.container.startsWith("self-hosted:")) { |
| 2123 | this.sql.exec("UPDATE jobs SET cancel_at = ?, cancel_reason = ? WHERE id = ?", at, reason, j.id); |
| 2124 | await this.stopContainer(j, true); |
| 2125 | this.repo.alarmBy(at + CANCEL_STOP_MS); |
| 2126 | continue; |
| 2127 | } |
| 2128 | await this.stopContainer(j); |
| 2129 | await this.completeJob(j.id, { conclusion: "cancelled", outputs: {}, failure: { message: j.cancel_reason ?? reason } }); |
| 2130 | } |
| 2131 | } |
| 2132 | |
| 2133 | async cancel(runId: string, caller: Caller): Promise<void> { |
| 2134 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 2135 | await this.cancelInternal(runId, `Cancelled by ${caller.login}`, true); |
| 2136 | } |
| 2137 | |
| 2138 | /** |
| 2139 | * A new attempt of a run, like GitHub's re-runs: `all` runs every job again; `failed` keeps the |
| 2140 | * jobs and matrix legs that succeeded, were skipped or were neutral (unless they depend on a job |
| 2141 | * that runs again) and runs the rest, with everything that depends on them. |
| 2142 | */ |
| 2143 | async rerun(runId: string, caller: Caller, scope: "all" | "failed" = "all"): Promise<RunInfo> { |
| 2144 | await this.mayRerun(caller); |
| 2145 | const old = this.runRow(runId); |
| 2146 | if (scope === "all") return this.newAttempt(old, caller, null); |
| 2147 | if (old.status !== "completed") throw new HttpError(409, "this run hasn't finished yet; its failed jobs can be re-run once it has"); |
| 2148 | const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", old.id).toArray(); |
| 2149 | const failed = jobs.filter((j) => !KEPT_CONCLUSIONS.has(j.conclusion ?? "")); |
| 2150 | if (!failed.length) throw new HttpError(409, "this run has no failed jobs to re-run"); |
| 2151 | return this.newAttempt(old, caller, await this.carryFor(old, jobs, new Set(failed.map((j) => j.id)))); |
| 2152 | } |
| 2153 | |
| 2154 | /** GitHub's "re-run a job": a new attempt that runs one job (one matrix leg) and every job that depends on it, keeping the rest. */ |
| 2155 | async rerunJob(jobId: string, caller: Caller): Promise<RunInfo> { |
| 2156 | await this.mayRerun(caller); |
| 2157 | const job = this.jobRow(jobId); |
| 2158 | const old = this.runRow(job.run_id); |
| 2159 | if (old.status !== "completed") throw new HttpError(409, "this run hasn't finished yet; its jobs can be re-run once it has"); |
| 2160 | const jobs = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", old.id).toArray(); |
| 2161 | return this.newAttempt(old, caller, await this.carryFor(old, jobs, new Set([job.id]))); |
| 2162 | } |
| 2163 | |
| 2164 | private async mayRerun(caller: Caller): Promise<void> { |
| 2165 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 2166 | if (caller.id.startsWith("job:")) throw new HttpError(403, "a workflow's GITHUB_TOKEN cannot re-run workflows"); |
| 2167 | await this.refuseIfBlocked("run"); |
| 2168 | } |
| 2169 | |
| 2170 | /** |
| 2171 | * What a new attempt of `old` keeps when the jobs `rerun` (ids of its job rows) run again: every |
| 2172 | * job that neither runs again nor depends (through `needs`, reusable workflow calls included) on |
| 2173 | * one that does, and the other matrix legs of a job only some of whose legs run again. A job the |
| 2174 | * earlier attempt never got to (the run was cancelled first) runs too. Null means run everything: |
| 2175 | * the workflow can't be read any more, or what failed was the workflow's set-up. |
| 2176 | */ |
| 2177 | private async carryFor(old: RunRow, jobs: JobRow[], rerun: Set<string>): Promise<RerunCarry | null> { |
| 2178 | let exp: Expanded; |
| 2179 | try { |
| 2180 | exp = await this.expanded(old); |
| 2181 | } catch { |
| 2182 | return null; |
| 2183 | } |
| 2184 | const defs = exp.workflow.jobs; |
| 2185 | if (jobs.some((j) => rerun.has(j.id) && !defs[j.job_key])) return null; |
| 2186 | const rows = new Map<string, JobRow[]>(); |
| 2187 | for (const j of jobs) rows.set(j.job_key, [...(rows.get(j.job_key) ?? []), j]); |
| 2188 | const seeded = (key: string) => !rows.get(key)?.length || rows.get(key)!.some((j) => rerun.has(j.id)); |
| 2189 | // The expanded workflow's `needs` already name a call's jobs, and what a called workflow's first jobs inherit from its caller. |
| 2190 | const memo = new Map<string, boolean>(); |
| 2191 | const runsAgain = (key: string): boolean => { |
| 2192 | const known = memo.get(key); |
| 2193 | if (known !== undefined) return known; |
| 2194 | memo.set(key, false); |
| 2195 | const v = seeded(key) || upstream(key); |
| 2196 | memo.set(key, v); |
| 2197 | return v; |
| 2198 | }; |
| 2199 | const upstream = (key: string): boolean => (defs[key]?.needs ?? []).some(runsAgain); |
| 2200 | const carry: RerunCarry = {}; |
| 2201 | for (const key of Object.keys(defs)) { |
| 2202 | if (upstream(key)) continue; |
| 2203 | const mine = rows.get(key) ?? []; |
| 2204 | if (!seeded(key)) { |
| 2205 | carry[key] = { all: mine.map((j) => j.id) }; |
| 2206 | continue; |
| 2207 | } |
| 2208 | const legs: Record<string, string[]> = {}; |
| 2209 | for (const j of mine) if (!rerun.has(j.id) && j.status === "completed") (legs[canonicalJson(JSON.parse(j.matrix))] ??= []).push(j.id); |
| 2210 | if (Object.keys(legs).length) carry[key] = { legs }; |
| 2211 | } |
| 2212 | return carry; |
| 2213 | } |
| 2214 | |
| 2215 | private async newAttempt(old: RunRow, caller: Caller, carry: RerunCarry | null): Promise<RunInfo> { |
| 2216 | const wf = (await this.loadWorkflows(old.sha)).find((w) => w.path === old.workflow_path)?.workflow; |
| 2217 | if (!wf) throw new HttpError(422, "workflow file no longer valid at this commit"); |
| 2218 | // Whoever started an earlier attempt still can't approve this one where an environment prevents self-review. |
| 2219 | const priorActors = [...new Set([...((old.prior_actors ? JSON.parse(old.prior_actors) : []) as string[]), (JSON.parse(old.actor) as Actor).id])]; |
| 2220 | const run = await this.createRun(wf, { |
| 2221 | event: old.event, |
| 2222 | ref: old.ref, |
| 2223 | sha: old.sha, |
| 2224 | headSha: old.head_sha ?? undefined, |
| 2225 | payload: JSON.parse(old.event_payload), |
| 2226 | caller, |
| 2227 | prNumber: old.pr_number ?? undefined, |
| 2228 | inputs: JSON.parse(old.inputs), |
| 2229 | title: old.display_title, |
| 2230 | // Re-running code an agent wrote keeps it acting as the agent, whoever re-runs it. |
| 2231 | agentOrigin: !!old.agent_origin, |
| 2232 | rerun: { runNumber: old.run_number, attempt: old.attempt + 1, priorActors, carry }, |
| 2233 | }); |
| 2234 | // Not started after all (Actions suspended meanwhile, or too many unfinished runs). |
| 2235 | if (!run) throw new HttpError(409, "the run was not started; try again later"); |
| 2236 | return run; |
| 2237 | } |
| 2238 | |
| 2239 | // ---------------------------------------------------------------- runner protocol |
| 2240 | |
| 2241 | /** Validate a job or sandbox token. Returns what it grants. */ |
| 2242 | async runnerAuth( |
| 2243 | token: string, |
| 2244 | ): Promise<{ kind: "job"; jobId: string; runId: string; role: Caller["role"]; permissions: JobPermissions; actorKind: Actor["kind"] } | { kind: "sandbox"; sandboxId: string; caller: Caller; taskPerson?: string } | null> { |
| 2245 | const parts = token.split("_"); |
| 2246 | if (parts.length < 4) return null; |
| 2247 | const hash = await sha256(token); |
| 2248 | if (parts[0] === "djt") { |
| 2249 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", parts[2]!).toArray()[0]; |
| 2250 | if (!j || j.token_hash !== hash || j.status === "completed") return null; |
| 2251 | // Bounded in time too: a job whose Workflow died never completes, and its token mustn't live on. |
| 2252 | const expires = jobTokenExpiry(j); |
| 2253 | if (expires === null || Date.now() >= expires) return null; |
| 2254 | const grant = this.repo.runtime.grantFor(j.id); |
| 2255 | // Code an agent wrote, pushed or opened a pull request with runs as that agent (repo/token-policy.ts). |
| 2256 | const run = this.sql.exec<{ agent_origin: number }>("SELECT agent_origin FROM runs WHERE id = ?", j.run_id).toArray()[0]; |
| 2257 | const actorKind: Actor["kind"] = run?.agent_origin ? "agent" : "system"; |
| 2258 | return { kind: "job", jobId: j.id, runId: j.run_id, role: grant.role, permissions: grant.permissions, actorKind }; |
| 2259 | } |
| 2260 | if (parts[0] === "dst") { |
| 2261 | 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]; |
| 2262 | if (!s || s.secret !== hash || s.status !== "running") return null; |
| 2263 | // A sandbox's token lasts a day at most, even if nobody closes the sandbox. |
| 2264 | if (Date.now() - Date.parse(s.created_at) > SANDBOX_TOKEN_TTL_MS) return null; |
| 2265 | return { kind: "sandbox", sandboxId: s.id, caller: JSON.parse(s.created_by) as Caller, ...(s.task_person ? { taskPerson: s.task_person } : {}) }; |
| 2266 | } |
| 2267 | return null; |
| 2268 | } |
| 2269 | |
| 2270 | private async authJob(jobId: string, token: string): Promise<JobRow> { |
| 2271 | const auth = await this.runnerAuth(token); |
| 2272 | if (!auth || auth.kind !== "job" || auth.jobId !== jobId) throw new HttpError(401, "invalid job token"); |
| 2273 | return this.jobRow(jobId); |
| 2274 | } |
| 2275 | |
| 2276 | /** Claims for an OIDC id token; side-effect free (unlike runnerSpec). */ |
| 2277 | async oidcClaims(jobId: string, token: string): Promise<{ allowed: boolean; github: Record<string, string> }> { |
| 2278 | const job = await this.authJob(jobId, token); |
| 2279 | if (job.status !== "in_progress") throw new HttpError(409, "job is not running"); |
| 2280 | const run = this.runRow(job.run_id); |
| 2281 | const exp = await this.expanded(run); |
| 2282 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| 2283 | const wf = exp.workflowOf.get(job.job_key)!; |
| 2284 | const allowed = this.repo.runtime.grantFor(jobId).permissions["id-token"] === "write"; |
| 2285 | const ctx = this.githubContext(run, localKey(job.job_key, exp.parentOf.get(job.job_key) ?? null), token) as Record<string, Json>; |
| 2286 | const github: Record<string, string> = {}; |
| 2287 | for (const [k, v] of Object.entries(ctx)) if (typeof v === "string" || typeof v === "number") github[k] = String(v); |
| 2288 | // Ids don't change when a name is reused, so a cloud trust policy can pin them; the environment is the one |
| 2289 | // whose protection rules this job passed (runnerSpec records its deployment). |
| 2290 | github.repository_owner_id = String(numericId(this.repo.meta.ownerId)); |
| 2291 | const actor = JSON.parse(run.actor) as Actor; |
| 2292 | if (actor.id) github.actor_id = String(numericId(actor.id)); |
| 2293 | const env = this.sql.exec<{ environment: string }>("SELECT environment FROM deployments WHERE job_id = ?", jobId).toArray()[0]; |
| 2294 | // The environment claim vouches for the job like the environment's secrets do: a job refused those |
| 2295 | // (a pull request ref, or agent code the policy keeps them from) gets the ref or pull request subject. |
| 2296 | if (env && this.repo.deploys.secretsAllowed(env.environment, run.ref, run.agent_origin ? "agent" : actor.kind)) github.environment = env.environment; |
| 2297 | // Lets a cloud trust policy refuse code an agent wrote, whatever ref or environment it ran on. |
| 2298 | github.agent_origin = run.agent_origin ? "true" : "false"; |
| 2299 | return { allowed, github }; |
| 2300 | } |
| 2301 | |
| 2302 | /** |
| 2303 | * What a job token may do with container packages (the `packages` permission). |
| 2304 | * A workflow without a permissions block gets the repository's default: read in |
| 2305 | * repositories with read-only tokens (the default for new ones), write in those that |
| 2306 | * keep GitHub's old permissive default. A permissions map that leaves `packages` out gets none. |
| 2307 | */ |
| 2308 | async packagesPermission(token: string): Promise<"none" | "read" | "write"> { |
| 2309 | const auth = await this.runnerAuth(token); |
| 2310 | return auth?.kind === "job" ? auth.permissions.packages : "none"; |
| 2311 | } |
| 2312 | |
| 2313 | /** |
| 2314 | * A pull_request run for a pull request from another repository (a fork): its jobs |
| 2315 | * never get write or secrets. Decided from the pull request's row, not the event payload. |
| 2316 | */ |
| 2317 | private untrusted(run: RunRow): boolean { |
| 2318 | return run.event === "pull_request" && this.repo.policy.isForkPull(run.pr_number); |
| 2319 | } |
| 2320 | |
| 2321 | /** |
| 2322 | * Turns away a hosted runner on an image other than the deployed one (see STALE_IMAGE_TRIES). |
| 2323 | * `deployed` is the image of the Worker that took the runner's request. After a deploy this |
| 2324 | * object can run the previous version for minutes, while the Worker is updated within seconds |
| 2325 | * and the containers come up on the new image: going by this object's own RUNNER_IMAGE then |
| 2326 | * turned away every runner on the new image until this object was updated too. |
| 2327 | */ |
| 2328 | private async refuseStaleRunner(job: JobRow, runner: JobRunnerInfo | null, deployed: string | null = null): Promise<void> { |
| 2329 | const want = deployed ?? this.env.RUNNER_IMAGE; |
| 2330 | if (!want || job.labels || runner?.image === want || (job.stale_tries ?? 0) >= STALE_IMAGE_TRIES) return; |
| 2331 | await this.waitForImage(job, `a runner started on ${runner?.image ?? "an older image"} while the deploy of ${want} was rolling out`); |
| 2332 | throw new HttpError(409, `this runner's image (${runner?.image ?? "unknown"}) was replaced by ${want}; the job starts again on a new runner`); |
| 2333 | } |
| 2334 | |
| 2335 | /** `deployed`: the runner image of the Worker that took the request (see refuseStaleRunner). */ |
| 2336 | async runnerSpec(jobId: string, token: string, runner: JobRunnerInfo | null = null, deployed: string | null = null): Promise<JobSpec> { |
| 2337 | const job = await this.authJob(jobId, token); |
| 2338 | await this.refuseStaleRunner(job, runner, deployed); |
| 2339 | const run = this.runRow(job.run_id); |
| 2340 | const exp = await this.expanded(run); |
| 2341 | const def: Job = exp.workflow.jobs[job.job_key]!; |
| 2342 | const wf = exp.workflowOf.get(job.job_key)!; |
| 2343 | const parent = exp.parentOf.get(job.job_key) ?? null; |
| 2344 | const short = localKey(job.job_key, parent); |
| 2345 | let scope: { inputs: Record<string, unknown>; secrets: Record<string, string> }; |
| 2346 | try { |
| 2347 | // A fork's code never sees the repository's secrets, as on GitHub, and an agent's only where an admin allows it. |
| 2348 | scope = await this.callContext(run, exp, parent, !this.untrusted(run) && (!run.agent_origin || this.repo.policy.settings().agentRunSecrets)); |
| 2349 | } catch (e) { |
| 2350 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } }); |
| 2351 | throw new HttpError(409, (e as Error).message); |
| 2352 | } |
| 2353 | const m = this.repo.meta; |
| 2354 | // Deployment environment (`environment:`): its branch policy, secrets and variables, and a deployment record. |
| 2355 | let environment: { secrets: Record<string, string>; vars: Record<string, string> } = { secrets: {}, vars: {} }; |
| 2356 | if (def.environment) { |
| 2357 | try { |
| 2358 | 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) }; |
| 2359 | // The name evaluated at launch, whose protection rules the job passed (older jobs evaluate it here). |
| 2360 | const name = job.environment ?? interpolate(def.environment.name, ctx); |
| 2361 | 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.`); |
| 2362 | const url = def.environment.url && !def.environment.url.includes("steps.") ? interpolate(def.environment.url, ctx) : undefined; |
| 2363 | const logUrl = `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`; |
| 2364 | // Code an agent wrote deploys as that agent (environment rules hold it to the agent's limits). |
| 2365 | const actor = JSON.parse(run.actor) as Actor; |
| 2366 | 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 }); |
| 2367 | } catch (e) { |
| 2368 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: (e as Error).message } }); |
| 2369 | throw new HttpError(409, (e as Error).message); |
| 2370 | } |
| 2371 | } |
| 2372 | const t = now(); |
| 2373 | this.sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ?, runner_info = ? WHERE id = ?", t, runner ? JSON.stringify(runner) : null, jobId); |
| 2374 | // Wake up at its deadline in case its runner freezes (enforceJobDeadlines). |
| 2375 | this.repo.alarmBy(this.jobDeadline({ started_at: t, timeout_minutes: job.timeout_minutes })!); |
| 2376 | 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}` }); |
| 2377 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 2378 | // Only the secrets the job's expressions name reach the runner, as on GitHub (plus the runner's own debug switches). |
| 2379 | const used = referencedSecrets(def, wf.env); |
| 2380 | if (used !== "all") for (const name of RUNNER_SECRETS) used.add(name); |
| 2381 | const secrets: Record<string, string> = { ...secretsFor({ ...scope.secrets, ...(this.untrusted(run) ? {} : environment.secrets) }, used), GITHUB_TOKEN: token, DIFFINERY_TOKEN: token }; |
| 2382 | // The job token can do what the effective `permissions:` allow (job level wins, like GitHub). |
| 2383 | const permissions = permissionsForJob(exp, job.job_key, { untrusted: this.untrusted(run), repoDefault: this.repo.policy.settings().defaultPermissions }); |
| 2384 | this.repo.runtime.grant(jobId, permissions); |
| 2385 | const idToken = permissions["id-token"] === "write"; |
| 2386 | return { |
| 2387 | ...(job.size_note && !job.labels ? { runnerSize: job.size_note } : {}), |
| 2388 | runId: run.id, |
| 2389 | jobId, |
| 2390 | jobKey: short, |
| 2391 | name: job.name, |
| 2392 | repo: { owner: m.owner, name: m.name, cloneUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, defaultBranch: m.defaultBranch }, |
| 2393 | sha: run.sha, |
| 2394 | ref: run.ref, |
| 2395 | workflowPath: run.workflow_path, |
| 2396 | github: this.githubContext(run, short, token), |
| 2397 | env: { ...wf.env, ...def.env }, |
| 2398 | defaults: { run: { ...wf.defaults.run, ...def.defaults.run } }, |
| 2399 | matrix: JSON.parse(job.matrix), |
| 2400 | strategy: JSON.parse(job.strategy), |
| 2401 | needs: await this.needsResults(run, exp, exp.needs.get(job.job_key) ?? [], parent), |
| 2402 | inputs: scope.inputs, |
| 2403 | vars: { ...this.variables(), ...environment.vars }, |
| 2404 | secrets, |
| 2405 | outputs: def.outputs, |
| 2406 | steps: def.steps.map((s) => ({ ...s, "timeout-minutes": s["timeout-minutes"] === undefined ? undefined : Number(s["timeout-minutes"]) })), |
| 2407 | container: def.container, |
| 2408 | services: def.services, |
| 2409 | timeoutMinutes: job.timeout_minutes, |
| 2410 | continueOnError: !!job.continue_on_error, |
| 2411 | idToken: !!idToken, |
| 2412 | apiUrl: this.env.PUBLIC_URL, |
| 2413 | token, |
| 2414 | runtime: await this.runtimeServices(run, job), |
| 2415 | }; |
| 2416 | } |
| 2417 | |
| 2418 | /** ACTIONS_RUNTIME_TOKEN and the service URLs; the token expires when the job would time out. */ |
| 2419 | private async runtimeServices(run: RunRow, job: JobRow): Promise<JobSpec["runtime"]> { |
| 2420 | if (!this.env.MASTER_KEY) return undefined; |
| 2421 | const origin = new URL(this.env.PUBLIC_URL).origin; |
| 2422 | const exp = Math.floor(Date.now() / 1000) + (job.timeout_minutes + 30) * 60; |
| 2423 | const { scopes } = this.repo.runtime.job(job.id); |
| 2424 | const token = await signRuntimeToken(this.env, origin, { repo: this.repo.meta.repoId, run: run.id, job: job.id, exp }, scopes); |
| 2425 | return { token, resultsUrl: `${origin}/`, cacheUrl: `${origin}/_runtime/`, runtimeUrl: `${origin}/_runtime/` }; |
| 2426 | } |
| 2427 | |
| 2428 | async runnerLogs(jobId: string, token: string, lines: LogLine[]): Promise<void> { |
| 2429 | const job = await this.authJob(jobId, token); |
| 2430 | let chars = job.log_chars ?? 0; |
| 2431 | if (chars > MAX_JOB_LOG) return; |
| 2432 | let seq = this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", jobId).one().n ?? -1; |
| 2433 | const out: Array<LogLine & { seq: number }> = []; |
| 2434 | for (const l of lines.slice(0, 5000)) { |
| 2435 | seq++; |
| 2436 | const t = String(l.t ?? "").slice(0, 40); |
| 2437 | const step = Number.isInteger(l.step) ? l.step : 0; |
| 2438 | let text = String(l.text ?? ""); |
| 2439 | if (text.length > 16384) text = `${text.slice(0, 16384)}…`; |
| 2440 | // Past the cap, one last line says so and the rest is dropped: the whole log is read into |
| 2441 | // memory when the job completes, to archive it. |
| 2442 | if (chars + text.length > MAX_JOB_LOG) { |
| 2443 | text = `The log is cut off here: a job's log is limited to ${MAX_JOB_LOG / 1048576} MiB of text.`; |
| 2444 | chars = MAX_JOB_LOG + 1; |
| 2445 | } else chars += text.length; |
| 2446 | this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, ?, ?)", jobId, seq, t, step, text); |
| 2447 | out.push({ t, step, text, seq }); |
| 2448 | if (chars > MAX_JOB_LOG) break; |
| 2449 | } |
| 2450 | this.sql.exec("UPDATE jobs SET log_chars = ? WHERE id = ?", chars, jobId); |
| 2451 | this.repo.broadcast(`job:${jobId}`, { action: "logs", lines: out }); |
| 2452 | } |
| 2453 | |
| 2454 | async runnerStep(jobId: string, token: string, index: number, report: StepReport): Promise<void> { |
| 2455 | const job = await this.authJob(jobId, token); |
| 2456 | const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>; |
| 2457 | const existing = steps.findIndex((s) => s.index === index); |
| 2458 | const next = { ...(existing >= 0 ? steps[existing] : {}), ...report, index }; |
| 2459 | if (existing >= 0) steps[existing] = next; |
| 2460 | else steps.push(next); |
| 2461 | steps.sort((a, b) => a.index - b.index); |
| 2462 | this.sql.exec("UPDATE jobs SET steps = ? WHERE id = ?", JSON.stringify(steps), jobId); |
| 2463 | this.repo.broadcast(`job:${jobId}`, { action: "step", step: next }); |
| 2464 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 2465 | } |
| 2466 | |
| 2467 | async runnerComplete(jobId: string, token: string, completion: JobCompletion, deployed: string | null = null): Promise<void> { |
| 2468 | const job = await this.authJob(jobId, token); |
| 2469 | // Cancelled while it ran: it ends here whatever stopped it, never starting again, and says who cancelled it. |
| 2470 | if (job.cancel_at) { |
| 2471 | const { replaced: _, ...rest } = completion; |
| 2472 | return this.completeJob(jobId, rest.conclusion === "cancelled" ? { ...rest, failure: { message: job.cancel_reason ?? "Cancelled" } } : rest); |
| 2473 | } |
| 2474 | if (completion.replaced && completion.conclusion !== "success" && !job.labels) { |
| 2475 | // The runner can't tell a rollout's SIGTERM from the platform's own; its image can: on the |
| 2476 | // deployed one, no deploy replaced it (the platform stopped or moved its container). |
| 2477 | const image = job.runner_info ? (JSON.parse(job.runner_info) as JobRunnerInfo).image : undefined; |
| 2478 | const want = deployed ?? this.env.RUNNER_IMAGE; |
| 2479 | return this.runnerReplaced(job, image && want && image === want ? "stopped" : "replaced"); |
| 2480 | } |
| 2481 | await this.completeJob(jobId, completion); |
| 2482 | } |
| 2483 | |
| 2484 | /** |
| 2485 | * The platform replaced a hosted job's runner while the job ran (a deploy's container rollout, or |
| 2486 | * the platform moving the container), or stopped its container without warning (`how` "stopped", |
| 2487 | * see containerStopped). Its stop signal reaches the running step too, so the job can't finish there. It starts again from the beginning on a new runner, as if it had just been |
| 2488 | * queued, up to MAX_RUNNER_RESTARTS times, when running its steps again can't repeat anything |
| 2489 | * outside the run: no step but checkout had started, or the job could only read (see `readOnly`). |
| 2490 | * Otherwise, or past the limit, the job fails and says why. |
| 2491 | */ |
| 2492 | private async runnerReplaced(job: JobRow, how: "replaced" | "stopped" = "replaced"): Promise<void> { |
| 2493 | const what = how === "replaced" ? "The platform replaced this job's runner with a new version" : "The platform stopped this job's container"; |
| 2494 | const run = this.runRow(job.run_id); |
| 2495 | const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>; |
| 2496 | let exp: Expanded | null = null; |
| 2497 | try { |
| 2498 | exp = await this.expanded(run); |
| 2499 | } catch (e) { |
| 2500 | console.error("could not read the job's steps", e); |
| 2501 | } |
| 2502 | const defs = exp?.workflow.jobs[job.job_key]?.steps ?? []; |
| 2503 | const ran = steps.filter((s) => s.startedAt && !/^actions\/checkout@/.test(defs[s.index]?.uses ?? "")); |
| 2504 | // The step the platform stopped: the last one that started. |
| 2505 | const stopped = ran.at(-1); |
| 2506 | const restarts = (job.restarts ?? 0) + 1; |
| 2507 | const repeatable = !stopped || (exp !== null && this.readOnly(run, job, exp)); |
| 2508 | if (!repeatable || restarts > MAX_RUNNER_RESTARTS) { |
| 2509 | const why = !repeatable |
| 2510 | ? `while step "${stopped!.name}" ran. Diffinery doesn't start a job again by itself once its steps have run if it could change something outside the run (it uses secrets, a deployment environment, or a token that can write), since running them twice could repeat what they did` |
| 2511 | : `${stopped ? "after its steps had started" : "before the job's steps ran"}, ${MAX_RUNNER_RESTARTS + 1} times`; |
| 2512 | await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message: `${what} ${why}. Re-run the job.` } }); |
| 2513 | return; |
| 2514 | } |
| 2515 | const t = now(); |
| 2516 | this.sql.exec( |
| 2517 | "UPDATE jobs SET status = 'queued', container = NULL, token_hash = NULL, started_at = NULL, steps = '[]', outputs = '{}', failure = NULL, summary = NULL, runner_info = NULL, log_chars = 0, restarts = ? WHERE id = ?", |
| 2518 | restarts, |
| 2519 | job.id, |
| 2520 | ); |
| 2521 | // The log starts over, with a line that says why (its seq keeps counting up, for live readers). |
| 2522 | const when = stopped ? `while step "${stopped.name}" ran. The job can only read (no secrets, deployment environment or token that can write), so` : "before the job's steps ran."; |
| 2523 | const text = `##[warning]${what} ${when} Diffinery started the job again from the beginning on a new runner (restart ${restarts} of ${MAX_RUNNER_RESTARTS}).`; |
| 2524 | const seq = (this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", job.id).one().n ?? -1) + 1; |
| 2525 | this.sql.exec("DELETE FROM logs WHERE job_id = ?", job.id); |
| 2526 | this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, -1, ?)", job.id, seq, t, text); |
| 2527 | this.sql.exec("UPDATE jobs SET log_chars = ? WHERE id = ?", text.length, job.id); |
| 2528 | // What the stopped attempt uploaded goes too: the new attempt uploads its own, under the same names. |
| 2529 | await this.repo.runtime.sweepJob(job.id, { restart: true }).catch((e) => console.error("runtime sweep failed", e)); |
| 2530 | await endCompute(this.env, job.id).catch((e: unknown) => console.error("compute usage failed", e)); |
| 2531 | 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) }); |
| 2532 | this.repo.broadcast(`job:${job.id}`, { action: "restarted" }); |
| 2533 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId: job.id }); |
| 2534 | try { |
| 2535 | const instance = await this.env.RUN_WORKFLOW.get(run.id); |
| 2536 | await instance.sendEvent({ type: `job-${job.id}`, payload: { retryInSeconds: 10 } }); |
| 2537 | } catch (e) { |
| 2538 | console.error("could not ask the run's workflow to start the job again", e); |
| 2539 | await this.completeJob(job.id, { conclusion: "failure", outputs: {}, failure: { message: `${what}, and Diffinery could not start the job again. Re-run the job.` } }); |
| 2540 | } |
| 2541 | } |
| 2542 | |
| 2543 | /** |
| 2544 | * Whether the job could only read, so running it again from the beginning can't repeat anything |
| 2545 | * outside the run: its steps name no secrets (its token aside), it deploys to no environment, and |
| 2546 | * its token's effective `permissions:` write nothing (an OIDC token counts as a write). What it |
| 2547 | * left in the run itself (log, check, uploads, caches) is reset when it starts again. |
| 2548 | */ |
| 2549 | private readOnly(run: RunRow, job: JobRow, exp: Expanded): boolean { |
| 2550 | try { |
| 2551 | const def = exp.workflow.jobs[job.job_key]; |
| 2552 | const wf = exp.workflowOf.get(job.job_key); |
| 2553 | if (!def || !wf || def.environment) return false; |
| 2554 | const used = referencedSecrets(def, wf.env); |
| 2555 | if (used === "all" || [...used].some((name) => name !== "GITHUB_TOKEN" && !RUNNER_SECRETS.includes(name))) return false; |
| 2556 | const permissions = permissionsForJob(exp, job.job_key, { untrusted: this.untrusted(run), repoDefault: this.repo.policy.settings().defaultPermissions }); |
| 2557 | return Object.values(permissions).every((access) => access !== "write"); |
| 2558 | } catch (e) { |
| 2559 | console.error("could not tell whether the job only reads", e); |
| 2560 | return false; |
| 2561 | } |
| 2562 | } |
| 2563 | |
| 2564 | private async completeJob(jobId: string, completion: JobCompletion, override?: "timed_out"): Promise<void> { |
| 2565 | const job = this.jobRow(jobId); |
| 2566 | if (job.status === "completed") return; |
| 2567 | const gated = job.status === "approval"; |
| 2568 | const run = this.runRow(job.run_id); |
| 2569 | const t = now(); |
| 2570 | const conclusion = override ?? completion.conclusion; |
| 2571 | this.sql.exec( |
| 2572 | "UPDATE jobs SET status = 'completed', conclusion = ?, completed_at = ?, outputs = ?, failure = ?, summary = ?, token_hash = NULL, started_at = COALESCE(started_at, ?), usage = COALESCE(?, usage) WHERE id = ?", |
| 2573 | conclusion, |
| 2574 | t, |
| 2575 | JSON.stringify(completion.outputs ?? {}), |
| 2576 | completion.failure ? JSON.stringify(completion.failure) : null, |
| 2577 | completion.summary ?? null, |
| 2578 | t, |
| 2579 | job.labels ? null : cleanUsage(completion.usage), |
| 2580 | jobId, |
| 2581 | ); |
| 2582 | // The run's last job: check later that the workflow finished the run (repairStuckRuns). |
| 2583 | if (!this.sql.exec("SELECT 1 FROM jobs WHERE run_id = ? AND status != 'completed' LIMIT 1", run.id).toArray().length) this.repo.alarmBy(Date.parse(t) + STUCK_RUN_MS); |
| 2584 | this.repo.deploys.jobCompleted(jobId, conclusion); |
| 2585 | const m = this.repo.meta; |
| 2586 | this.repo.upsertCheck({ |
| 2587 | sha: run.head_sha ?? run.sha, |
| 2588 | name: this.checkName(run, job), |
| 2589 | status: "completed", |
| 2590 | conclusion: conclusion === "timed_out" ? "timed_out" : conclusion, |
| 2591 | runId: run.id, |
| 2592 | jobId, |
| 2593 | detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`, |
| 2594 | output: completion.failure ? { title: "Failed", summary: completion.failure.message } : { title: conclusion }, |
| 2595 | }); |
| 2596 | // Archive logs to R2 and free the Durable Object's storage. |
| 2597 | const lines = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? ORDER BY seq", jobId).toArray(); |
| 2598 | if (lines.length) { |
| 2599 | await this.env.BUCKET.put(this.logKey(jobId), JSON.stringify(lines), { httpMetadata: { contentType: "application/json" } }); |
| 2600 | this.sql.exec("DELETE FROM logs WHERE job_id = ?", jobId); |
| 2601 | } |
| 2602 | // Uploads the job left unfinished stop counting against the budgets, and their bytes go. |
| 2603 | await this.repo.runtime.sweepJob(jobId).catch((e) => console.error("runtime sweep failed", e)); |
| 2604 | this.repo.broadcast(`job:${jobId}`, { action: "completed", conclusion }); |
| 2605 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 2606 | |
| 2607 | // fail-fast: cancel unfinished siblings of a failed matrix job (unless the failed leg may continue on error). |
| 2608 | 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(); |
| 2609 | const strategy = JSON.parse(job.strategy || "{}") as { "fail-fast"?: boolean }; |
| 2610 | if ((conclusion === "failure" || conclusion === "timed_out") && !job.continue_on_error && strategy["fail-fast"] !== false && siblings.length) { |
| 2611 | for (const s of siblings) { |
| 2612 | await this.stopContainer(s); |
| 2613 | await this.completeJob(s.id, { conclusion: "cancelled", outputs: {}, failure: { message: `Cancelled because ${job.name} failed (fail-fast).` } }); |
| 2614 | } |
| 2615 | } else { |
| 2616 | // A leg that never started (max-parallel), not one back in line after getting no machine or |
| 2617 | // being restarted: the run's workflow starts that one again (placementFailed, the safe restart). |
| 2618 | const queued = siblings.find((s) => s.status === "queued" && !s.placement_tries && !s.restarts); |
| 2619 | if (queued) await this.launch(queued.id); |
| 2620 | } |
| 2621 | // An ephemeral runner is done once its one job is. |
| 2622 | if (job.runner_id) this.sql.exec("DELETE FROM runners WHERE id = ? AND ephemeral = 1 AND token_hash IS NULL", job.runner_id); |
| 2623 | // A hosted job hands its slot to the oldest job waiting for one (the repository's or the account's). |
| 2624 | if (!job.labels && job.container) { |
| 2625 | await endCompute(this.env, jobId).catch((e: unknown) => console.error("compute usage failed", e)); |
| 2626 | const held = this.sql.exec<{ id: string }>("SELECT id FROM jobs WHERE fleet_held > 0 AND status = 'queued' ORDER BY rowid LIMIT 1").toArray()[0]; |
| 2627 | if (held) await this.launch(held.id); |
| 2628 | } |
| 2629 | // A job that held its concurrency group hands it to the oldest pending job (of any run). |
| 2630 | if (job.concurrency_group && (job.status === "waiting" || job.status === "in_progress")) { |
| 2631 | 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]; |
| 2632 | if (next) await this.launch(next.id); |
| 2633 | } |
| 2634 | try { |
| 2635 | const instance = await this.env.RUN_WORKFLOW.get(run.id); |
| 2636 | await instance.sendEvent({ type: `job-${jobId}`, payload: { conclusion } }); |
| 2637 | } catch (e) { |
| 2638 | console.error("failed to notify workflow", e); |
| 2639 | } |
| 2640 | // The Workflow waits for a job held for approval on its own event. |
| 2641 | if (gated) { |
| 2642 | this.syncWaiting(run.id); |
| 2643 | await this.wakeGate(run.id, jobId); |
| 2644 | } |
| 2645 | } |
| 2646 | |
| 2647 | // ---------------------------------------------------------------- secrets and variables |
| 2648 | |
| 2649 | private async repoSecrets(): Promise<Record<string, string>> { |
| 2650 | const out: Record<string, string> = {}; |
| 2651 | for (const r of this.sql.exec<{ name: string; ciphertext: string }>("SELECT name, ciphertext FROM secrets").toArray()) { |
| 2652 | // Sealed with a previous MASTER_KEY: stored again with the current one (lib/crypto.ts), keeping updated_at, which is when a person set it. |
| 2653 | out[r.name] = await decryptAndReseal(this.env, `repo:${this.repo.meta.repoId}`, r.ciphertext, (sealed) => |
| 2654 | this.sql.exec("UPDATE secrets SET ciphertext = ? WHERE name = ? AND ciphertext = ?", sealed, r.name, r.ciphertext), |
| 2655 | ); |
| 2656 | } |
| 2657 | return out; |
| 2658 | } |
| 2659 | |
| 2660 | private async orgSecrets(): Promise<Record<string, string>> { |
| 2661 | const m = this.repo.meta; |
| 2662 | if (m.ownerKind !== "org") return {}; |
| 2663 | // Only the secrets whose repository selection includes this repository (visibility all, private or selected). |
| 2664 | const rows = await this.env.DB.prepare( |
| 2665 | `SELECT s.name, s.ciphertext FROM org_secrets s WHERE s.org_id = ?1 AND (s.visibility = 'all' |
| 2666 | OR (s.visibility = 'private' AND EXISTS (SELECT 1 FROM repos r WHERE r.id = ?2 AND r.visibility != 'public')) |
| 2667 | 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)))`, |
| 2668 | ) |
| 2669 | .bind(m.ownerId, m.repoId) |
| 2670 | .all<{ name: string; ciphertext: string }>(); |
| 2671 | const out: Record<string, string> = {}; |
| 2672 | for (const r of rows.results) { |
| 2673 | out[r.name] = await decryptAndReseal(this.env, `org:${m.ownerId}`, r.ciphertext, (sealed) => |
| 2674 | this.env.DB.prepare("UPDATE org_secrets SET ciphertext = ? WHERE org_id = ? AND name = ? AND ciphertext = ?").bind(sealed, m.ownerId, r.name, r.ciphertext).run(), |
| 2675 | ); |
| 2676 | } |
| 2677 | return out; |
| 2678 | } |
| 2679 | |
| 2680 | async setSecret(name: string, value: string): Promise<void> { |
| 2681 | if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || /^GITHUB_/i.test(name)) throw new HttpError(422, "invalid secret name"); |
| 2682 | this.sql.exec("INSERT OR REPLACE INTO secrets (name, ciphertext, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), await encryptSecret(this.env, `repo:${this.repo.meta.repoId}`, value), now()); |
| 2683 | } |
| 2684 | |
| 2685 | deleteSecret(name: string): void { |
| 2686 | this.sql.exec("DELETE FROM secrets WHERE name = ?", name.toUpperCase()); |
| 2687 | } |
| 2688 | |
| 2689 | secretNames(): Array<{ name: string; updatedAt: string }> { |
| 2690 | 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 })); |
| 2691 | } |
| 2692 | |
| 2693 | setVariable(name: string, value: string | null): void { |
| 2694 | if (value === null) this.sql.exec("DELETE FROM variables WHERE name = ?", name.toUpperCase()); |
| 2695 | else this.sql.exec("INSERT OR REPLACE INTO variables (name, value, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), value, now()); |
| 2696 | } |
| 2697 | |
| 2698 | /** Variables for `vars.*`: the organization's, overridden by the repository's. `withOrg = false` lists only the repository's own. */ |
| 2699 | variables(withOrg = true): Record<string, string> { |
| 2700 | const own = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM variables").toArray().map((r) => [r.name, r.value])); |
| 2701 | if (!withOrg) return own; |
| 2702 | const org = Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM org_variables").toArray().map((r) => [r.name, r.value])); |
| 2703 | return { ...org, ...own }; |
| 2704 | } |
| 2705 | |
| 2706 | // ---------------------------------------------------------------- cache and artifacts |
| 2707 | |
| 2708 | /** |
| 2709 | * An older runner's cache (the runner API before the toolkit's cache service). |
| 2710 | * Entries are stored as `<ref>\n<key>` and scoped like the cache service's |
| 2711 | * (repo/runtime.ts): a job saves to its own ref and restores from its own ref, |
| 2712 | * then the pull request's base, then the default branch, so one branch or pull |
| 2713 | * request never sees another's caches. Entries saved before scoping (no ref) |
| 2714 | * are never restored, so they go a week after their last use. |
| 2715 | */ |
| 2716 | async cacheLookup(jobId: string, token: string, key: string, restoreKeys: string[]): Promise<{ key: string; stored: string } | null> { |
| 2717 | await this.authJob(jobId, token); |
| 2718 | const { scopes } = this.repo.runtime.job(jobId); |
| 2719 | // An entry whose object is gone is forgotten, and the lookup moves on. |
| 2720 | for (let attempt = 0; attempt < 5; attempt++) { |
| 2721 | const hit = this.legacyCacheMatch(scopes, key, restoreKeys); |
| 2722 | if (!hit) return null; |
| 2723 | if (!(await headFirst(this.env.BUCKET, this.cacheR2Keys(hit)))) { |
| 2724 | this.sql.exec("DELETE FROM caches WHERE key = ?", hit); |
| 2725 | continue; |
| 2726 | } |
| 2727 | this.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", now(), hit); |
| 2728 | return { key: hit.slice(hit.indexOf("\n") + 1), stored: hit }; |
| 2729 | } |
| 2730 | return null; |
| 2731 | } |
| 2732 | |
| 2733 | /** In each scope in turn: the exact key, then the newest entry for each restore key prefix. */ |
| 2734 | private legacyCacheMatch(scopes: string[], key: string, restoreKeys: string[]): string | null { |
| 2735 | for (const scope of scopes) { |
| 2736 | const exact = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE key = ?", `${scope}\n${key}`).toArray()[0]; |
| 2737 | if (exact) return exact.key; |
| 2738 | for (const prefix of restoreKeys) { |
| 2739 | const p = `${scope}\n${prefix}`; |
| 2740 | 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]; |
| 2741 | if (r) return r.key; |
| 2742 | } |
| 2743 | } |
| 2744 | return null; |
| 2745 | } |
| 2746 | |
| 2747 | /** Where a job saves `key` (its own ref's scope): the stored key and its R2 keys. */ |
| 2748 | async cacheSaveKeys(jobId: string, token: string, key: string): Promise<{ stored: string; r2Keys: string[] }> { |
| 2749 | await this.authJob(jobId, token); |
| 2750 | const stored = `${this.repo.runtime.job(jobId).scopes[0]}\n${key}`; |
| 2751 | return { stored, r2Keys: this.cacheR2Keys(stored) }; |
| 2752 | } |
| 2753 | |
| 2754 | /** The R2 keys of entry `stored`, when the job may read its scope. */ |
| 2755 | async cacheReadKeys(jobId: string, token: string, stored: string): Promise<string[] | null> { |
| 2756 | await this.authJob(jobId, token); |
| 2757 | const nl = stored.indexOf("\n"); |
| 2758 | if (nl < 0 || !this.repo.runtime.job(jobId).scopes.includes(stored.slice(0, nl))) return null; |
| 2759 | return this.cacheR2Keys(stored); |
| 2760 | } |
| 2761 | |
| 2762 | /** Where an older runner's cache entry is stored: the current key, then the old layout's. */ |
| 2763 | cacheR2Keys(key: string): string[] { |
| 2764 | return legacyCacheKeys(this.repo.meta.repoId, key); |
| 2765 | } |
| 2766 | |
| 2767 | /** Records a saved entry; returns the R2 keys of the entries it evicted, for the caller to delete. */ |
| 2768 | async cacheSaved(jobId: string, token: string, key: string, size: number): Promise<string[]> { |
| 2769 | const { stored } = await this.cacheSaveKeys(jobId, token, key); |
| 2770 | this.sql.exec("INSERT OR REPLACE INTO caches (key, size, created_at, last_used) VALUES (?, ?, ?, ?)", stored, size, now(), now()); |
| 2771 | // Like the cache service (repo/runtime.ts): entries unused for a week go, then the |
| 2772 | // least recently used ones above 10 GB per repository. |
| 2773 | 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); |
| 2774 | for (const k of stale) this.sql.exec("DELETE FROM caches WHERE key = ?", k); |
| 2775 | const evicted = stale.flatMap((k) => this.cacheR2Keys(k)); |
| 2776 | let total = this.sql.exec<{ n: number | null }>("SELECT sum(size) AS n FROM caches").one().n ?? 0; |
| 2777 | while (total > CACHE_LIMIT_BYTES) { |
| 2778 | const old = this.sql.exec<{ key: string; size: number }>("SELECT key, size FROM caches ORDER BY last_used LIMIT 1").one(); |
| 2779 | this.sql.exec("DELETE FROM caches WHERE key = ?", old.key); |
| 2780 | evicted.push(...this.cacheR2Keys(old.key)); |
| 2781 | total -= old.size; |
| 2782 | } |
| 2783 | return evicted; |
| 2784 | } |
| 2785 | |
| 2786 | /** |
| 2787 | * Where job `jobId` (authenticated by its token) stores artifact `name` of its run, and how many bytes |
| 2788 | * it may store there (what is left of the artifact quota). Refused when the run |
| 2789 | * already has an artifact of that name from the artifact service. |
| 2790 | */ |
| 2791 | async artifactUploadKeys(jobId: string, token: string, name: string): Promise<{ r2Keys: string[]; room: number }> { |
| 2792 | const job = await this.authJob(jobId, token); |
| 2793 | this.repo.runtime.legacyArtifactCheck(job.run_id, name); |
| 2794 | return { r2Keys: this.artifactR2Keys(job.run_id, name), room: await this.repo.runtime.artifactRoom() }; |
| 2795 | } |
| 2796 | |
| 2797 | /** Records an older runner's artifact, stored at its legacy key, with the artifact service's (repo/runtime.ts). */ |
| 2798 | async artifactSaved(jobId: string, token: string, name: string, size: number): Promise<{ id: number }> { |
| 2799 | const job = await this.authJob(jobId, token); |
| 2800 | return { id: this.repo.runtime.legacyArtifactSaved(job.run_id, jobId, name, size).id }; |
| 2801 | } |
| 2802 | |
| 2803 | /** Where job `jobId` (authenticated by its token) reads artifact `name` of its run. */ |
| 2804 | async artifactKeysFor(jobId: string, token: string, name: string): Promise<string[]> { |
| 2805 | const job = await this.authJob(jobId, token); |
| 2806 | return this.artifactR2Keys(job.run_id, name); |
| 2807 | } |
| 2808 | |
| 2809 | /** Where an older runner's artifact is stored: the current key, then the old layout's. */ |
| 2810 | artifactR2Keys(runId: string, name: string): string[] { |
| 2811 | return legacyArtifactKeys(this.repo.meta.repoId, runId, name); |
| 2812 | } |
| 2813 | |
| 2814 | /** |
| 2815 | * A run's artifacts: those uploaded through the artifact service (zip files, or |
| 2816 | * single unarchived files) and older runners' tarballs. |
| 2817 | */ |
| 2818 | artifacts(runId: string): Array<{ name: string; size: number; createdAt: string; r2Keys: string[]; id: number; contentType: string; fileName: string }> { |
| 2819 | runId = this.runRow(runId).id; |
| 2820 | return this.repo.runtime |
| 2821 | .runArtifacts(runId) |
| 2822 | .map((a) => ({ |
| 2823 | name: a.name, |
| 2824 | size: a.size, |
| 2825 | createdAt: a.createdAt, |
| 2826 | r2Keys: artifactObjectKeys(this.repo.meta.repoId, a), |
| 2827 | id: a.id, |
| 2828 | contentType: a.mimeType, |
| 2829 | fileName: artifactFileName(a), |
| 2830 | })) |
| 2831 | .sort((a, b) => a.name.localeCompare(b.name)); |
| 2832 | } |
| 2833 | |
| 2834 | // ---------------------------------------------------------------- sandboxes for agents |
| 2835 | |
| 2836 | /** |
| 2837 | * `task` is set only by agentic/ (never from a request): the sandbox of an agent task, started for |
| 2838 | * the agent `caller` by the person `task.personId`. No public sandbox call reaches it. An agent |
| 2839 | * review's sandbox is `readOnly`: stored with a `read` caller, its git credential can't push. |
| 2840 | * A web terminal's sandbox (`kind: "terminal"`) is a task sandbox one person works in directly. |
| 2841 | */ |
| 2842 | async createSandbox(caller: Caller, ref?: string, task?: { personId: string; readOnly?: boolean; kind?: "task" | "terminal" }): Promise<{ id: string; ref: string }> { |
| 2843 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to create a sandbox"); |
| 2844 | // A review's sandbox acts with read access only: its git credential can clone but never push. |
| 2845 | if (task?.readOnly) caller = { ...caller, role: "read" }; |
| 2846 | if (!task) { |
| 2847 | const since = new Date(Date.now() - SANDBOX_TOKEN_TTL_MS).toISOString(); |
| 2848 | const open = this.sql |
| 2849 | .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) |
| 2850 | .one().n; |
| 2851 | if (open >= MAX_OPEN_SANDBOXES) throw new HttpError(429, `you have ${open} open sandboxes in this repository, the most allowed; close one first`); |
| 2852 | } |
| 2853 | await this.refuseIfBlocked("sandbox", false); |
| 2854 | // The account that owns the repository pays for sandboxes too (lib/quotas.ts). |
| 2855 | const quota = await mayStart(this.env, this.repo.meta.ownerId, "sandbox"); |
| 2856 | if (!quota.ok) throw new HttpError(429, quota.message); |
| 2857 | const sha = await this.repo.resolve(ref ?? "HEAD"); |
| 2858 | if (!sha) throw new HttpError(422, "unknown ref"); |
| 2859 | const id = newId(); |
| 2860 | const m = this.repo.meta; |
| 2861 | const token = `dst_${m.repoId}_${id}_${randomToken(24)}`; |
| 2862 | const secret = randomToken(32); |
| 2863 | this.sql.exec( |
| 2864 | "INSERT INTO sandboxes (id, secret, created_by, ref, created_at, status, kind, task_person) VALUES (?, ?, ?, ?, ?, 'running', ?, ?)", |
| 2865 | id, |
| 2866 | await sha256(token), |
| 2867 | JSON.stringify(caller), |
| 2868 | ref ?? m.defaultBranch, |
| 2869 | now(), |
| 2870 | task ? (task.kind ?? "task") : "api", |
| 2871 | task?.personId ?? null, |
| 2872 | ); |
| 2873 | await startCompute(this.env, { id, accountId: m.ownerId, repoId: m.repoId, kind: "sandbox" }); |
| 2874 | const stub = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)); |
| 2875 | try { |
| 2876 | await stub.boot({ |
| 2877 | DIFFINERY_CLONE_URL: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 2878 | DIFFINERY_REF: ref ?? m.defaultBranch, |
| 2879 | DIFFINERY_JOB_TOKEN: token, |
| 2880 | DIFFINERY_SANDBOX_SECRET: secret, |
| 2881 | DIFFINERY_API_URL: this.env.PUBLIC_URL, |
| 2882 | GIT_AUTHOR_NAME: caller.login, |
| 2883 | GIT_AUTHOR_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 2884 | GIT_COMMITTER_NAME: caller.login, |
| 2885 | GIT_COMMITTER_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 2886 | }, secret, id, { id, accountId: m.ownerId, repoId: m.repoId }, await egressFor(this.env, m.ownerId)); |
| 2887 | } catch (e) { |
| 2888 | console.error("sandbox boot failed", e); |
| 2889 | this.sql.exec("UPDATE sandboxes SET status = 'failed' WHERE id = ?", id); |
| 2890 | await endCompute(this.env, id).catch(() => {}); |
| 2891 | await stub.shutdown().catch(() => {}); |
| 2892 | throw new HttpError(503, "the sandbox container did not start; try again"); |
| 2893 | } |
| 2894 | return { id, ref: ref ?? m.defaultBranch }; |
| 2895 | } |
| 2896 | |
| 2897 | getSandbox(id: string, caller: Caller): { id: string; ref: string; status: string; createdAt: string } { |
| 2898 | 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]; |
| 2899 | // A task's sandbox holds a credential for the person's model access, and a terminal's may hold their |
| 2900 | // provider sign-in: nobody reaches either through the sandbox API, admins included. |
| 2901 | if (!s || s.kind !== "api") throw new HttpError(404, "sandbox not found"); |
| 2902 | const owner = JSON.parse(s.created_by) as Caller; |
| 2903 | if (owner.id !== caller.id && !atLeast(caller.role, "admin")) throw new HttpError(403, "not your sandbox"); |
| 2904 | return { id: s.id, ref: s.ref, status: s.status, createdAt: s.created_at }; |
| 2905 | } |
| 2906 | |
| 2907 | async closeSandbox(id: string, caller: Caller): Promise<void> { |
| 2908 | this.getSandbox(id, caller); |
| 2909 | await this.endSandbox(id); |
| 2910 | } |
| 2911 | |
| 2912 | /** Close a sandbox: its token stops working, its time stops counting and its container stops. */ |
| 2913 | private async endSandbox(id: string): Promise<void> { |
| 2914 | this.sql.exec("UPDATE sandboxes SET status = 'closed' WHERE id = ?", id); |
| 2915 | await endCompute(this.env, id); |
| 2916 | await this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)).shutdown(); |
| 2917 | } |
| 2918 | |
| 2919 | /** Close an agent task's or a web terminal's sandbox (agentic/), which revokes its git credential. */ |
| 2920 | async closeTaskSandbox(id: string): Promise<void> { |
| 2921 | const s = this.sql.exec<{ kind: string }>("SELECT kind FROM sandboxes WHERE id = ?", id).toArray()[0]; |
| 2922 | if (s?.kind !== "task" && s?.kind !== "terminal") return; |
| 2923 | await this.endSandbox(id); |
| 2924 | } |
| 2925 | |
| 2926 | // ---------------------------------------------------------------- kill switches (lib/killswitch.ts) |
| 2927 | |
| 2928 | /** |
| 2929 | * Why `work` may not start here now, or null. One D1 read. A failed read lets work start |
| 2930 | * (`failOpen`), as quotas do for jobs: a D1 outage shouldn't stop every repository's CI. |
| 2931 | */ |
| 2932 | private async blocked(work: Work, failOpen = true): Promise<string | null> { |
| 2933 | try { |
| 2934 | return blockedBy(await killSwitches(this.env, this.repo.meta.repoId), work); |
| 2935 | } catch (e) { |
| 2936 | if (!failOpen) throw e; |
| 2937 | console.error("kill switch check failed", e); |
| 2938 | return null; |
| 2939 | } |
| 2940 | } |
| 2941 | |
| 2942 | /** Refuse (403, with the reason) what someone asked to start while a kill switch stops it. */ |
| 2943 | async refuseIfBlocked(work: Work, failOpen = true): Promise<void> { |
| 2944 | const block = await this.blocked(work, failOpen); |
| 2945 | if (block) throw new HttpError(403, block); |
| 2946 | } |
| 2947 | |
| 2948 | /** |
| 2949 | * A site administrator just suspended Actions here or for the owner (`hostedOnly` false): every |
| 2950 | * unfinished run is cancelled and every open sandbox closed. Or paused hosted compute on the |
| 2951 | * instance (`hostedOnly`): running hosted jobs fail and open sandboxes close; self-hosted jobs go on. |
| 2952 | * `message` says why. Jobs still queued fail when they would start (launch). |
| 2953 | */ |
| 2954 | async stopActions(o: { hostedOnly: boolean; message: string }): Promise<{ runs: number; jobs: number; sandboxes: number }> { |
| 2955 | let runs = 0; |
| 2956 | let jobs = 0; |
| 2957 | let sandboxes = 0; |
| 2958 | if (o.hostedOnly) { |
| 2959 | for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE labels IS NULL AND container IS NOT NULL AND status IN ('waiting', 'in_progress') ORDER BY rowid").toArray()) { |
| 2960 | await this.stopContainer(j); |
| 2961 | await this.completeJob(j.id, { conclusion: "failure", outputs: {}, failure: { message: `The job was stopped: ${o.message}.` } }); |
| 2962 | jobs++; |
| 2963 | } |
| 2964 | } else { |
| 2965 | for (const r of this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE status != 'completed' ORDER BY rowid").toArray()) { |
| 2966 | jobs += this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM jobs WHERE run_id = ? AND status != 'completed'", r.id).one().n; |
| 2967 | await this.cancelInternal(r.id, `Cancelled: ${o.message}.`); |
| 2968 | runs++; |
| 2969 | } |
| 2970 | } |
| 2971 | for (const s of this.sql.exec<{ id: string }>("SELECT id FROM sandboxes WHERE status = 'running'").toArray()) { |
| 2972 | await this.endSandbox(s.id).catch((e: unknown) => console.error("closing a sandbox failed", e)); |
| 2973 | sandboxes++; |
| 2974 | } |
| 2975 | return { runs, jobs, sandboxes }; |
| 2976 | } |
| 2977 | } |
| 2978 | |
| 2979 | /** |
| 2980 | * One minute of a wait timer: a real minute, except in the end-to-end tests (WAIT_TIMER_MINUTE_MS). The setting |
| 2981 | * shortens a protection rule, so it counts only on a Worker that runs on this machine (auth/dev.ts): a deployed |
| 2982 | * one ignores it. |
| 2983 | */ |
| 2984 | export function timerMinute(env: Pick<Env, "WAIT_TIMER_MINUTE_MS" | "PUBLIC_URL" | "ACCESS_AUD">): number { |
| 2985 | if (!env.WAIT_TIMER_MINUTE_MS || !localOnly(env)) return 60_000; |
| 2986 | const ms = Number(env.WAIT_TIMER_MINUTE_MS); |
| 2987 | return Number.isInteger(ms) && ms > 0 && ms <= 60_000 ? ms : 60_000; |
| 2988 | } |