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