Skip to content
Sign up

flagon-io/g1t

public
0
2,509 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

GitHub Actions on g1t, part two: running workflows1//! A run's life: made, its jobs waiting on the jobs they need, each job
2//! skipped or expanded into its matrix and queued, started in a sandbox
3//! when its workspace has room, reporting its steps and logs as it goes,
4//! and finished; the run finishes with its last job.
5
6use g1t_actions::events::{RunInfo, runner_context};
7use g1t_actions::expr::{self, Scope, Status};
8use g1t_actions::matrix;
9use g1t_actions::workflow::{self, Workflow};
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look10use g1t_contracts::access::Capability;
Merge branch 'worktree-agent-a3abfcce648e87dca'11use g1t_contracts::actions::{JobCallArgs, RunActionArgs, RunApproval, StartJobArgs, WorkflowRun};
12use g1t_contracts::identity::{CreateJobTokenArgs, CreatedAccessToken, RevokeJobTokensArgs};
GitHub Actions on g1t, part two: running workflows13use g1t_contracts::repos::{Repo, RepoPath};
14use g1t_contracts::time::rfc3339;
15use g1t_contracts::{FailureCode, Outcome, new_id};
16use g1t_kit::now_ms;
17use g1t_secrets::{random_hex, same, sha256_hex};
18use serde::Deserialize;
19use serde_json::{Map, Value, json};
20use worker::Result;
21
Merge branch 'worktree-agent-a3abfcce648e87dca'22use crate::protection::Gate;
GitHub Actions on g1t, part two: running workflows23use crate::sync::WorkflowRow;
Fast pages, required checks on the branch, self-hosted runners, honest incidents24use crate::{Actions, Count, MAX_TIMEOUT_MINUTES, RUNNING_PER_WORKSPACE, SELF_HOSTED_MAX_TIMEOUT_MINUTES, SILENT_MS, SITE, check, fail, optional, repo_path};
25use g1t_contracts::runners::{Wanted, waiting_reason};
GitHub Actions on g1t, part two: running workflows26
27/// The most log one job keeps, in bytes; past it, the log says so and stops.
28const MAX_LOG_BYTES: usize = 4 * 1024 * 1024;
29/// The most a single log report may add.
30const MAX_CHUNK_BYTES: usize = 256 * 1024;
31const MAX_ANNOTATIONS: usize = 50;
Actions: keep every run attempt, re-run one job, graceful cancel, summaries32/// The most one step's job summary keeps, in bytes, and how many steps of a
33/// job may have one, as on GitHub.
34pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
35pub(crate) const MAX_SUMMARIES: u32 = 20;
36
37/// Whether a step's summary of `adding` bytes is kept: a job keeps up to
38/// 20 steps' summaries (`steps` it has, `mine` this step's bytes so far)
39/// of up to 1 MiB each.
40pub(crate) fn summary_fits(steps: u32, mine: Option<usize>, adding: usize) -> bool {
41 if adding == 0 {
42 return false;
43 }
44 match mine {
45 Some(used) => used + adding <= MAX_SUMMARY_BYTES,
46 None => steps < MAX_SUMMARIES && adding <= MAX_SUMMARY_BYTES,
47 }
48}
GitHub Actions on g1t, part two: running workflows49
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look50/// The repository whose unfinished runs `event` stops, by id: one deleted,
51/// or archived (not unarchived).
52pub fn stops_runs(event: &g1t_contracts::events::Event) -> Option<String> {
53 let stops = match event.kind.as_str() {
54 "repo.deleted" => true,
55 "repo.archived" => event.data["archived"].as_bool() != Some(false),
56 _ => false,
57 };
58 if !stops {
59 return None;
60 }
61 event.repo_id.clone().or_else(|| event.data["repoId"].as_str().map(str::to_owned))
62}
63
GitHub Actions on g1t, part two: running workflows64pub struct NewRun {
65 pub repo: Repo,
66 pub path: String,
67 pub source: String,
68 pub workflow: Workflow,
69 pub info: RunInfo,
70 pub action: Option<String>,
71 pub pull: Option<u32>,
72 pub title: String,
73 pub inputs: Map<String, Value>,
74 pub event_key: String,
75 pub actor_id: Option<String>,
76 pub actor: Option<String>,
77 pub trusted: bool,
Merge branch 'worktree-agent-a3abfcce648e87dca'78 /// Why the run waits for someone with the Write role to approve it
79 /// before anything starts: a pull request from outside (protection.rs).
80 pub approval: Option<String>,
GitHub Actions on g1t, part two: running workflows81}
82
83#[derive(Clone, Deserialize)]
84pub struct RunRow {
85 pub id: String,
86 pub workflow_id: String,
87 pub repo_id: String,
88 pub repo: String,
89 pub path: String,
90 pub name: String,
91 pub title: String,
92 pub number: u64,
93 pub attempt: u64,
94 pub event: String,
95 pub action: Option<String>,
96 pub git_ref: String,
97 pub sha: String,
98 pub pull: Option<u32>,
99 pub status: String,
100 pub conclusion: Option<String>,
101 pub error: Option<String>,
102 pub actor: Option<String>,
103 pub actor_id: Option<String>,
104 pub source: String,
105 pub info: String,
106 pub inputs: String,
107 pub trusted: u32,
108 pub concurrency_group: Option<String>,
109 pub created_at: String,
110 pub started_at: Option<String>,
111 pub finished_at: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'112 /// JSON `RunApproval`, for a run that needed approval (migration 0006).
113 #[serde(default)]
114 pub approval: Option<String>,
115 /// Whether its concurrency group cancels what it replaces.
116 #[serde(default)]
117 pub cancel_in_progress: u32,
Actions: keep every run attempt, re-run one job, graceful cancel, summaries118 /// Who started the current attempt, once it is re-run (migration 0008).
119 #[serde(default)]
120 pub triggering_actor: Option<String>,
121 /// The current attempt runs with debug logging.
122 #[serde(default)]
123 pub debug: u32,
GitHub Actions on g1t, part two: running workflows124}
125
126impl RunRow {
Merge branch 'worktree-agent-a3abfcce648e87dca'127 pub fn approval(&self) -> Option<RunApproval> {
128 self.approval.as_deref().and_then(|text| serde_json::from_str(text).ok())
129 }
130
GitHub Actions on g1t, part two: running workflows131 pub fn info(&self) -> RunInfo {
132 let mut info: RunInfo = serde_json::from_str(&self.info).unwrap_or_default();
133 info.run_id = self.id.clone();
134 info.run_number = self.number;
135 info.run_attempt = self.attempt;
136 info.workflow_path = self.path.clone();
137 if info.workflow.is_empty() {
138 info.workflow = self.name.clone();
139 }
140 info
141 }
142
143 pub fn inputs(&self) -> Map<String, Value> {
144 serde_json::from_str(&self.inputs).unwrap_or_default()
145 }
146
147 pub fn summary(&self) -> WorkflowRun {
148 WorkflowRun {
149 id: self.id.clone(),
150 workflow_id: self.workflow_id.clone(),
151 path: self.path.clone(),
152 name: self.name.clone(),
153 title: self.title.clone(),
154 number: self.number,
155 attempt: self.attempt,
156 event: self.event.clone(),
157 git_ref: self.git_ref.clone(),
158 sha: self.sha.clone(),
159 pull: self.pull,
160 status: self.status.clone(),
161 conclusion: self.conclusion.clone(),
162 error: self.error.clone(),
163 actor: self.actor.clone(),
164 created_at: self.created_at.clone(),
165 started_at: self.started_at.clone(),
166 finished_at: self.finished_at.clone(),
167 }
168 }
169}
170
171#[derive(Clone, Deserialize)]
172pub struct JobRow {
173 pub id: String,
174 pub run_id: String,
175 pub repo_id: String,
176 pub namespace: String,
177 pub key: String,
178 pub ordinal: u32,
179 pub name: String,
180 pub needs: String,
181 pub matrix: Option<String>,
182 pub status: String,
183 pub conclusion: Option<String>,
184 pub steps: String,
185 pub annotations: String,
186 pub outputs: String,
187 pub reason: Option<String>,
188 pub token_hash: Option<String>,
189 pub timeout_minutes: u32,
190 pub continue_on_error: u32,
191 pub max_parallel: Option<u32>,
Actions: reusable workflows in the repository192 /// Set for a job that calls a reusable workflow, and for that
193 /// workflow's jobs (see migration 0002).
194 pub call: Option<String>,
GitHub Actions on g1t, part two: running workflows195 pub seen_at: Option<String>,
196 pub started_at: Option<String>,
197 pub finished_at: Option<String>,
Fast pages, required checks on the branch, self-hosted runners, honest incidents198 /// For a job whose `runs-on` names self-hosted runners: what it asks
199 /// for, as a JSON array (see `g1t_contracts::runners::Wanted`), when it
200 /// started waiting, and the runner that took it (migration 0004).
201 #[serde(default)]
202 pub labels: Option<String>,
203 #[serde(default)]
204 pub queued_at: Option<String>,
205 #[serde(default)]
206 pub runner_id: Option<String>,
207 #[serde(default)]
208 pub runner_name: Option<String>,
Merge branch 'worktree-agent-a3abfcce648e87dca'209 /// The environment it names, read when its needs were done; a job its
210 /// rules hold is `pending` (migration 0006).
211 #[serde(default)]
212 pub environment: Option<String>,
213 /// Its own `concurrency` group, and whether that cancels what it replaces.
214 #[serde(default)]
215 pub concurrency_group: Option<String>,
216 #[serde(default)]
217 pub cancel_in_progress: u32,
Actions: keep every run attempt, re-run one job, graceful cancel, summaries218 /// When it was told to stop, while it runs its cleanup steps
219 /// (migration 0008).
220 #[serde(default)]
221 pub cancel_requested_at: Option<String>,
222}
223
224/// How long a cancelled job has to run its `if: always()` and `cancelled()`
225/// steps and its post steps before it is stopped outright, as on GitHub.
226pub const CANCEL_GRACE_MS: u64 = 5 * 60 * 1000;
227
228/// Which jobs a re-run runs again.
229#[derive(Clone, Copy, Debug, PartialEq)]
230pub enum Rerun<'a> {
231 All,
232 /// Those that did not succeed.
233 Failed,
234 /// One job's key.
235 Job(&'a str),
236}
237
238/// The keys a re-run runs again, in `order`: those `which` names and,
239/// transitively, every key that needs one of them. `needs` gives a key's
240/// needs; `succeeded` whether every job of a key succeeded.
241pub fn rerun_keys(order: &[&str], needs: impl Fn(&str) -> Vec<String>, succeeded: impl Fn(&str) -> bool, which: Rerun) -> Vec<String> {
242 let mut again: Vec<String> = Vec::new();
243 for key in order {
244 let named = match which {
245 Rerun::All => true,
246 Rerun::Failed => !succeeded(key),
247 Rerun::Job(job) => *key == job,
248 };
249 if named || needs(key).iter().any(|need| again.contains(need)) {
250 again.push((*key).to_owned());
251 }
252 }
253 again
254}
255
256/// The key under `jobs:` a job belongs to: a called workflow's jobs
257/// (`build/test`) run again with the job that calls it (`build`).
258pub fn top_key(key: &str) -> &str {
259 key.split('/').next().unwrap_or(key)
260}
261
262/// Debug logging for a job, as a re-run with it turned on gives GitHub's:
263/// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and
264/// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of
265/// those names (`::debug::` lines shown, and how each step's `if` read).
266pub fn debug_logging(variables: &mut Map<String, Value>, runner: &mut Value) {
267 variables.insert("RUNNER_DEBUG".into(), json!("1"));
268 variables.insert("ACTIONS_STEP_DEBUG".into(), json!("true"));
269 variables.insert("ACTIONS_RUNNER_DEBUG".into(), json!("true"));
270 runner["debug"] = json!("1");
GitHub Actions on g1t, part two: running workflows271}
272
273impl JobRow {
274 pub fn needs(&self) -> Vec<String> {
275 serde_json::from_str(&self.needs).unwrap_or_default()
276 }
Actions: reusable workflows in the repository277
278 pub fn call(&self) -> Option<Value> {
279 self.call.as_deref().and_then(|call| serde_json::from_str(call).ok())
280 }
281
282 /// For a job of a called workflow: that workflow, the job's own id in
283 /// it, and the job.
284 pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> {
285 let call = self.call().filter(|call| call["role"] == "callee")?;
286 let called = workflow::parse(call["source"].as_str()?).ok()?;
287 let job = called.jobs.iter().find(|job| call["job"].as_str() == Some(job.id.as_str()))?.clone();
288 Some((called, job, call))
289 }
GitHub Actions on g1t, part two: running workflows290}
291
Actions: reusable workflows in the repository292/// A job to decide on: its key, its definition, and what it needs, as
293/// (name in `needs`, key of the jobs).
294type Unit = (String, workflow::Job, Vec<(String, String)>);
295
296/// How deep reusable workflows may call one another, as on GitHub.
297const MAX_CALL_DEPTH: u64 = 4;
298
GitHub Actions on g1t, part two: running workflows299/// What the jobs of one key came to, for `needs.<key>`.
300fn key_result(rows: &[&JobRow]) -> &'static str {
301 let failed = |row: &&&JobRow| row.conclusion.as_deref() == Some("failure") && row.continue_on_error == 0;
302 if rows.iter().any(|row| failed(&row)) {
303 "failure"
304 } else if rows.iter().any(|row| row.conclusion.as_deref() == Some("cancelled")) {
305 "cancelled"
306 } else if rows.iter().all(|row| row.conclusion.as_deref() == Some("skipped")) {
307 "skipped"
308 } else {
309 "success"
310 }
311}
312
Fast pages, required checks on the branch, self-hosted runners, honest incidents313/// Whether any job before `key` failed: one it needs, or one those need,
314/// however far back. A job after a skipped one still sees the failure
315/// that skipped it, as GitHub's failure() does. `needs_of`: each key's
316/// needs, as keys; `failed`: whether a key's jobs came to a failure.
317fn ancestor_failed(needs_of: &std::collections::HashMap<&str, Vec<&str>>, key: &str, failed: impl Fn(&str) -> bool) -> bool {
318 let mut seen = std::collections::HashSet::new();
319 let mut stack: Vec<&str> = needs_of.get(key).cloned().unwrap_or_default();
320 while let Some(next) = stack.pop() {
321 if !seen.insert(next) {
322 continue;
323 }
324 if failed(next) {
325 return true;
326 }
327 stack.extend(needs_of.get(next).into_iter().flatten().copied());
328 }
329 false
330}
331
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97332/// A job's `environment:`, as deployments read it: its name, the address in
333/// `environment.url` once its expressions are filled in from the run, and
334/// whether the job deploys to it. A job with `deployment: false` only reads
335/// the environment's secrets and variables, and makes no deployment.
336#[derive(Clone, Debug, Default, PartialEq)]
337pub(crate) struct JobEnvironment {
338 pub(crate) name: String,
339 pub(crate) url: Option<String>,
340 pub(crate) deploys: bool,
341}
342
343/// The environment `environment:` names, as written in `raw` (a job), with
344/// `contexts` to fill in expressions: null when it names none, or only
345/// through an expression this cannot read before the job runs.
346pub(crate) fn environment_of(raw: &Value, contexts: &Map<String, Value>) -> Option<JobEnvironment> {
347 let scope = Scope { contexts, status: Status::Success, hash_files: None };
348 let plain = |value: &Value| -> Option<String> {
349 let text = match value {
350 Value::String(text) if text.contains("${{") => expr::interpolate_value(value, &scope).ok().map(|v| expr::to_text(&v))?,
351 Value::String(text) => text.clone(),
352 _ => return None,
353 };
354 let text = text.trim().to_owned();
355 (!text.is_empty()).then_some(text)
356 };
357 match raw.get("environment")? {
358 name @ Value::String(_) => Some(JobEnvironment { name: plain(name)?, url: None, deploys: true }),
359 Value::Object(env) => {
360 let name = plain(env.get("name")?)?;
361 let url = env
362 .get("url")
363 .and_then(plain)
364 .filter(|url| url.starts_with("https://") || url.starts_with("http://"));
365 let deploys = !matches!(env.get("deployment"), Some(Value::Bool(false)))
366 && !matches!(env.get("deployment"), Some(Value::String(text)) if text.trim() == "false");
367 Some(JobEnvironment { name, url, deploys })
368 }
369 _ => None,
370 }
371}
372
373/// The contexts a job's `runs-on` and `environment` are read with before it
374/// runs: the run's `github`, its inputs, and the job's matrix.
375fn start_contexts(run: &RunRow, job: &JobRow, key: &str) -> Map<String, Value> {
376 let mut contexts = Map::new();
377 contexts.insert("github".into(), run.info().context(key, "", run.action.as_deref()));
378 contexts.insert("inputs".into(), Value::Object(run.inputs()));
379 contexts.insert("matrix".into(), job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or_else(|| json!({})));
380 contexts
381}
382
383/// The job as its workflow (or the workflow it calls) defines it.
384fn job_spec(run: &RunRow, job: &JobRow) -> Option<workflow::Job> {
385 match job.callee() {
386 Some((_, spec, _)) => Some(spec),
387 None => workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)),
388 }
389}
390
Merge branch 'worktree-agent-a3abfcce648e87dca'391/// The environment a job deploys to, if it deploys: by the name read when
392/// its needs were done (which may have needed their outputs), else as
393/// its `environment:` reads now.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97394pub(crate) fn deploys_to(run: &RunRow, job: &JobRow) -> Option<JobEnvironment> {
395 let spec = job_spec(run, job)?;
Merge branch 'worktree-agent-a3abfcce648e87dca'396 let read = environment_of(&spec.raw, &start_contexts(run, job, &spec.id));
397 let env = match (read, &job.environment) {
398 (Some(env), Some(name)) => Some(JobEnvironment { name: name.clone(), ..env }),
399 (Some(env), None) => Some(env),
400 (None, Some(name)) => {
401 let deployment = spec.raw.get("environment").and_then(|env| env.get("deployment"));
402 let deploys = !matches!(deployment, Some(Value::Bool(false)))
403 && !matches!(deployment, Some(Value::String(text)) if text.trim() == "false");
404 Some(JobEnvironment { name: name.clone(), url: None, deploys })
405 }
406 (None, None) => None,
407 };
408 env.filter(|env| env.deploys)
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97409}
410
Fast pages, required checks on the branch, self-hosted runners, honest incidents411/// What the runner is told about a job it starts, from its workflow: the
412/// environment it names plainly, and the machine its `runs-on` asks for
413/// (`instance_for`; none for the standard one).
414#[derive(Default)]
415struct StartDetails {
416 environment: Option<String>,
417 instance: Option<String>,
418}
419
420fn start_details(run: &RunRow, job: &JobRow) -> StartDetails {
421 let spec = match job.callee() {
422 Some((_, spec, _)) => spec,
423 None => match workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)) {
424 Some(spec) => spec,
425 None => return StartDetails::default(),
426 },
427 };
Merge branch 'worktree-agent-a3abfcce648e87dca'428 // As read when its needs were done, an expression's included.
429 let environment = job.environment.clone();
Fast pages, required checks on the branch, self-hosted runners, honest incidents430 // `runs-on` as the job was queued with: its matrix and the run's
431 // inputs. A label that needs more than those is the standard machine.
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97432 let contexts = start_contexts(run, job, &spec.id);
Fast pages, required checks on the branch, self-hosted runners, honest incidents433 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
434 let labels: Vec<String> = match expr::interpolate_value(&spec.runs_on, &scope).unwrap_or(Value::Null) {
435 Value::String(label) => vec![label],
436 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
437 Value::Object(given) => match given.get("labels") {
438 Some(Value::Array(labels)) => labels.iter().map(expr::to_text).collect(),
439 Some(label) => vec![expr::to_text(label)],
440 None => Vec::new(),
441 },
442 _ => Vec::new(),
443 };
444 let instance = g1t_contracts::actions::instance_for(&labels);
445 StartDetails {
446 environment,
447 instance: (instance != g1t_contracts::actions::STANDARD_INSTANCE).then(|| instance.label.to_owned()),
448 }
449}
450
GitHub Actions on g1t, part two: running workflows451fn now() -> String {
452 rfc3339(now_ms())
453}
454
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97455/// How a finished run went for one environment, from the conclusions of
456/// the jobs that deploy to it: failed if any failed, an error if any was
457/// cancelled, else a success; nothing when none ran (all skipped).
458pub(crate) fn deployment_outcome(conclusions: &[Option<String>]) -> Option<&'static str> {
459 let ran: Vec<&str> = conclusions.iter().flatten().map(String::as_str).filter(|c| *c != "skipped").collect();
460 if ran.is_empty() {
461 return None;
462 }
463 if ran.iter().any(|c| *c == "failure" || *c == "timed_out") {
464 return Some("failure");
465 }
466 if ran.contains(&"cancelled") {
467 return Some("error");
468 }
469 Some("success")
470}
471
GitHub Actions on g1t, part two: running workflows472impl Actions {
473 pub async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
474 self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<RunRow>(None).await
475 }
476
477 pub async fn job_rows(&self, run_id: &str) -> Result<Vec<JobRow>> {
478 self.db
479 .prepare("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid")
480 .bind(&[run_id.into()])?
481 .all()
482 .await?
483 .results::<JobRow>()
484 }
485
486 pub async fn run_summary(&self, id: &str) -> Result<Outcome<WorkflowRun>> {
487 Ok(match self.run_row(id).await? {
488 Some(row) => Outcome::Ok(row.summary()),
489 None => fail(FailureCode::NotFound, "No such run."),
490 })
491 }
492
493 /// The contexts every expression outside a job's steps may use.
494 fn base_contexts(run: &RunRow, vars: &Map<String, Value>, job: &str) -> Map<String, Value> {
495 let mut contexts = Map::new();
496 contexts.insert("github".into(), run.info().context(job, "", run.action.as_deref()));
497 contexts.insert("inputs".into(), Value::Object(run.inputs()));
498 contexts.insert("vars".into(), Value::Object(vars.clone()));
499 contexts.insert("needs".into(), json!({}));
500 contexts.insert("runner".into(), runner_context());
501 contexts
502 }
503
504 /// Makes a run and its jobs, and starts what can start. `None` when the
505 /// event already started this workflow.
506 pub async fn create_run(&self, new: NewRun) -> Result<Option<String>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look507 // Nothing starts on an archived repository. A deleted one is never
508 // found to start anything on.
509 if new.repo.archived() {
510 return Ok(None);
511 }
GitHub Actions on g1t, part two: running workflows512 let workflow_row = self.workflow_row(&new.repo, &new.path, &new.workflow.display_name(&new.path), &new.source).await?;
513 let numbered = self
514 .db
515 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
516 .bind(&[workflow_row.id.as_str().into()])?
517 .first::<Count>(None)
518 .await?
519 .map_or(1, |count| count.n);
520 let id = new_id("run", now_ms());
521 let mut info = new.info.clone();
522 info.workflow = new.workflow.display_name(&new.path);
523 info.workflow_path = new.path.clone();
524 info.run_id = id.clone();
525 info.run_number = u64::from(numbered);
Secrets and variables: one list, rows per environment, for workflows and deployments526 let vars = self
527 .variables_for(&new.repo.id, &format!("{}/{}", new.repo.namespace, new.repo.name), None, new.trusted)
528 .await?;
GitHub Actions on g1t, part two: running workflows529
530 // run-name and the concurrency group read github, inputs and vars.
531 let mut contexts = Map::new();
532 contexts.insert("github".into(), info.context("", "", new.action.as_deref()));
533 contexts.insert("inputs".into(), Value::Object(new.inputs.clone()));
534 contexts.insert("vars".into(), Value::Object(vars));
535 let scope = Scope {
536 contexts: &contexts,
537 status: Status::Success,
538 hash_files: None,
539 };
540 let title = new
541 .workflow
542 .run_name
543 .as_deref()
544 .and_then(|run_name| expr::interpolate(run_name, &scope).ok())
545 .filter(|title| !title.trim().is_empty())
546 .unwrap_or(new.title.clone());
547 let group = new.workflow.concurrency.as_ref().and_then(|c| expr::interpolate(&c.group, &scope).ok());
548 let cancel_in_progress = new
549 .workflow
550 .concurrency
551 .as_ref()
552 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
553 .is_some_and(|value| expr::truthy(&value));
554
Merge branch 'worktree-agent-a3abfcce648e87dca'555 let approval = new.approval.as_ref().map(|reason| RunApproval { state: "required".into(), reason: reason.clone(), approved_by: None });
GitHub Actions on g1t, part two: running workflows556 let inserted = self
557 .db
558 .prepare(
559 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, action, git_ref, sha,
Merge branch 'worktree-agent-a3abfcce648e87dca'560 pull, status, actor, actor_id, source, info, inputs, trusted, concurrency_group, event_key, created_at,
561 approval, cancel_in_progress)
562 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
GitHub Actions on g1t, part two: running workflows563 )
564 .bind(&[
565 id.as_str().into(),
566 workflow_row.id.as_str().into(),
567 new.repo.id.as_str().into(),
568 format!("{}/{}", new.repo.namespace, new.repo.name).into(),
569 new.path.as_str().into(),
570 info.workflow.as_str().into(),
571 title.as_str().into(),
572 numbered.into(),
573 info.event_name.as_str().into(),
574 optional(new.action.as_deref()),
575 info.git_ref.as_str().into(),
576 info.sha.as_str().into(),
577 new.pull.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
Merge branch 'worktree-agent-a3abfcce648e87dca'578 if approval.is_some() { "action_required" } else { "queued" }.into(),
GitHub Actions on g1t, part two: running workflows579 optional(new.actor.as_deref()),
580 optional(new.actor_id.as_deref()),
581 new.source.as_str().into(),
582 serde_json::to_string(&info)?.into(),
583 serde_json::to_string(&new.inputs)?.into(),
584 u32::from(new.trusted).into(),
585 optional(group.as_deref()),
586 new.event_key.as_str().into(),
587 now().into(),
Merge branch 'worktree-agent-a3abfcce648e87dca'588 approval.as_ref().map(serde_json::to_string).transpose()?.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
589 u32::from(cancel_in_progress).into(),
GitHub Actions on g1t, part two: running workflows590 ])?
591 .first::<Value>(None)
592 .await?;
593 if inserted.is_none() {
594 return Ok(None);
595 }
596 if let Some(run) = self.run_row(&id).await? {
597 self.report_pending(&run).await?;
598 }
599
600 // Every job, waiting; each is expanded when the jobs it needs are done.
601 let mut statements = Vec::new();
602 for job in &new.workflow.jobs {
603 statements.push(
604 self.db
605 .prepare(
606 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting')",
607 )
608 .bind(&[
609 new_id("job", now_ms()).into(),
610 id.as_str().into(),
611 new.repo.id.as_str().into(),
612 new.repo.namespace.as_str().into(),
613 job.id.as_str().into(),
614 job.name.clone().filter(|n| !expr::has_expression(n)).unwrap_or(job.id.clone()).into(),
615 serde_json::to_string(&job.needs)?.into(),
616 ])?,
617 );
618 }
619 self.db.batch(statements).await?;
620
Merge branch 'worktree-agent-a3abfcce648e87dca'621 // A pull request's run from outside waits to be approved; it joins
622 // its concurrency group once it is (protection.rs, approve_run).
623 if approval.is_some() {
624 return Ok(Some(id));
625 }
626 if let Some(run) = self.run_row(&id).await? {
627 self.enter_group(&run).await?;
628 }
629 Ok(Some(id))
630 }
631
632 /// A run about to start: one run at a time per concurrency group, as on
633 /// GitHub. A newer run replaces one that waits in the group, and with
634 /// `cancel-in-progress` the one that runs; otherwise it waits as
635 /// `pending` for the one that runs. Then it moves along.
636 pub(crate) async fn enter_group(&self, run: &RunRow) -> Result<()> {
637 if let Some(group) = &run.concurrency_group {
638 let cancel_in_progress = run.cancel_in_progress != 0;
GitHub Actions on g1t, part two: running workflows639 let others = self
640 .db
Merge branch 'worktree-agent-a3abfcce648e87dca'641 .prepare(
642 "SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND id != ? AND status NOT IN ('completed', 'action_required') ORDER BY id",
643 )
644 .bind(&[run.repo_id.as_str().into(), group.as_str().into(), run.id.as_str().into()])?
GitHub Actions on g1t, part two: running workflows645 .all()
646 .await?
647 .results::<RunRow>()?;
648 for other in &others {
649 if cancel_in_progress || other.status == "pending" {
650 // A newer run replaces a waiting one, as on GitHub.
Actions: keep every run attempt, re-run one job, graceful cancel, summaries651 self.cancel_run(other, "A newer run in the same concurrency group replaced it.", false).await?;
GitHub Actions on g1t, part two: running workflows652 }
653 }
654 if !cancel_in_progress && others.iter().any(|other| other.status != "pending") {
655 self.db
656 .prepare("UPDATE runs SET status = 'pending' WHERE id = ?")
Merge branch 'worktree-agent-a3abfcce648e87dca'657 .bind(&[run.id.as_str().into()])?
GitHub Actions on g1t, part two: running workflows658 .run()
659 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'660 return Ok(());
GitHub Actions on g1t, part two: running workflows661 }
662 }
Merge branch 'worktree-agent-a3abfcce648e87dca'663 self.advance(&run.id).await
GitHub Actions on g1t, part two: running workflows664 }
665
666 /// A run that could not start, such as for a workflow file that does not read.
667 #[allow(clippy::too_many_arguments)]
668 pub async fn record_failed_run(
669 &self,
670 row: &WorkflowRow,
671 git_ref: &str,
672 sha: &str,
673 event_key: &str,
674 actor_id: Option<&str>,
675 actor: &str,
676 problem: &str,
677 ) -> Result<()> {
678 let numbered = self
679 .db
680 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
681 .bind(&[row.id.as_str().into()])?
682 .first::<Count>(None)
683 .await?
684 .map_or(1, |count| count.n);
685 let at = now();
686 self.db
687 .prepare(
688 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, git_ref, sha, status,
689 conclusion, error, actor, actor_id, source, info, event_key, created_at, finished_at)
690 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'push', ?, ?, 'completed', 'failure', ?, ?, ?, ?, '{}', ?, ?, ?)",
691 )
692 .bind(&[
693 new_id("run", now_ms()).into(),
694 row.id.as_str().into(),
695 row.repo_id.as_str().into(),
696 row.repo.as_str().into(),
697 row.path.as_str().into(),
698 row.path.as_str().into(),
699 "Invalid workflow file".into(),
700 numbered.into(),
701 git_ref.into(),
702 sha.into(),
703 problem.into(),
704 actor.into(),
705 optional(actor_id),
706 row.source.as_str().into(),
707 event_key.into(),
708 at.as_str().into(),
709 at.as_str().into(),
710 ])?
711 .run()
712 .await?;
713 Ok(())
714 }
715
716 /// Moves a run along: jobs whose needs are done are decided on, and
717 /// jobs that can start are started.
718 pub async fn advance(&self, run_id: &str) -> Result<()> {
719 // Each pass may finish jobs (skipped ones), which may free others.
720 for _ in 0..20 {
721 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'722 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
GitHub Actions on g1t, part two: running workflows723 return Ok(());
724 }
725 let workflow = match workflow::parse(&run.source) {
726 Ok(workflow) => workflow,
727 Err(problem) => {
728 self.finish_run(&run, Some(&problem)).await?;
729 return Ok(());
730 }
731 };
732 let jobs = self.job_rows(run_id).await?;
733 let mut changed = false;
Actions: reusable workflows in the repository734 // What to decide on: the workflow's jobs, and the jobs of the
735 // workflows they call, each with its needs as (name, key).
736 let mut units: Vec<Unit> = workflow
737 .jobs
738 .iter()
739 .map(|job| (job.id.clone(), job.clone(), job.needs.iter().map(|n| (n.clone(), n.clone())).collect()))
740 .collect();
741 let mut seen = std::collections::HashSet::new();
742 for row in &jobs {
743 if !seen.insert(row.key.clone()) {
744 continue;
745 }
746 if let Some((_, job, call)) = row.callee() {
747 let parent = call["parent"].as_str().unwrap_or_default().to_owned();
748 let needs = job.needs.iter().map(|n| (n.clone(), format!("{parent}/{n}"))).collect();
749 units.push((row.key.clone(), job, needs));
750 }
751 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents752 // Each key's needs, as keys, to look back through for failure().
753 let needs_of: std::collections::HashMap<&str, Vec<&str>> =
754 units.iter().map(|(key, _, needs)| (key.as_str(), needs.iter().map(|(_, need)| need.as_str()).collect())).collect();
755 let key_failed = |key: &str| key_result(&jobs.iter().filter(|row| row.key == key).collect::<Vec<_>>()) == "failure";
Actions: reusable workflows in the repository756 for (key, job, needs) in &units {
757 let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect();
GitHub Actions on g1t, part two: running workflows758 if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting") {
759 continue;
760 }
761 let needed: Vec<(&String, Vec<&JobRow>)> =
Actions: reusable workflows in the repository762 needs.iter().map(|(name, need)| (name, jobs.iter().filter(|row| &row.key == need).collect())).collect();
GitHub Actions on g1t, part two: running workflows763 if !needed.iter().all(|(_, rows)| rows.iter().all(|row| row.status == "completed")) {
764 continue;
765 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents766 let failed_before = ancestor_failed(&needs_of, key, key_failed);
767 self.decide(&run, job, rows[0], &needed, failed_before).await?;
GitHub Actions on g1t, part two: running workflows768 changed = true;
769 }
Actions: reusable workflows in the repository770 // A job that called a workflow finishes with that workflow's jobs.
771 for row in jobs.iter().filter(|row| row.status == "calling") {
772 let children: Vec<&JobRow> = jobs
773 .iter()
774 .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(row.key.as_str())))
775 .collect();
776 if !children.is_empty() && children.iter().all(|child| child.status == "completed") {
777 self.finish_call(row, &children).await?;
778 changed = true;
779 }
780 }
GitHub Actions on g1t, part two: running workflows781 if !changed {
782 break;
783 }
784 }
785 self.start_queued().await?;
786 self.finish_if_done(run_id).await
787 }
788
789 /// Decides on one job whose needs are done: skip it, fail it, or expand
Fast pages, required checks on the branch, self-hosted runners, honest incidents790 /// it into its matrix and queue it. `failed_before`: whether any job
791 /// before it failed, however far back (`ancestor_failed`).
792 async fn decide(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, needed: &[(&String, Vec<&JobRow>)], failed_before: bool) -> Result<()> {
Secrets and variables: one list, rows per environment, for workflows and deployments793 let vars = self.variables_for(&run.repo_id, &run.repo, None, run.trusted != 0).await?;
GitHub Actions on g1t, part two: running workflows794 let mut contexts = Self::base_contexts(run, &vars, &job.id);
Actions: reusable workflows in the repository795 // A called workflow's jobs read the inputs they were called with.
796 let call = row.call();
797 let parent = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
798 if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") {
799 contexts.insert("inputs".into(), call["inputs"].clone());
800 }
GitHub Actions on g1t, part two: running workflows801 let mut needs = Map::new();
Fast pages, required checks on the branch, self-hosted runners, honest incidents802 let mut results = Vec::new();
GitHub Actions on g1t, part two: running workflows803 for (key, rows) in needed {
804 let result = key_result(rows);
805 let mut outputs = Map::new();
806 for row in rows {
807 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
808 outputs.extend(more);
809 }
810 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents811 results.push(result);
GitHub Actions on g1t, part two: running workflows812 needs.insert((*key).clone(), json!({ "result": result, "outputs": outputs }));
813 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents814 // As on GitHub: a need that was skipped makes success() false but
815 // not failure(); failure() is a failure anywhere before the job.
816 let status = expr::job_status(results, failed_before, run.conclusion.as_deref() == Some("cancelled"));
GitHub Actions on g1t, part two: running workflows817 contexts.insert("needs".into(), Value::Object(needs));
818 let scope = Scope {
819 contexts: &contexts,
820 status,
821 hash_files: None,
822 };
823 let condition = job.condition.as_deref().unwrap_or_default();
824 match expr::condition(condition, &scope) {
825 Ok(true) => {}
826 Ok(false) => return self.skip_job(row, None).await,
827 Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await,
828 }
Actions: reusable workflows in the repository829 if let Some(uses) = &job.uses {
830 return self.call_workflow(run, job, row, uses, &scope).await;
GitHub Actions on g1t, part two: running workflows831 }
832
833 // Its matrix, which may come from a needed job's outputs.
834 let combinations = match &job.matrix {
835 None => vec![Map::new()],
836 Some(matrix) => {
837 let value = match expr::interpolate_value(matrix, &scope) {
838 Ok(value) => value,
839 Err(problem) => return self.fail_job(row, &format!("Its matrix does not read: {problem}")).await,
840 };
841 match matrix::expand(&value) {
842 Ok(combinations) if !combinations.is_empty() => combinations,
843 Ok(_) => return self.fail_job(row, "Its matrix makes no jobs.").await,
844 Err(problem) => return self.fail_job(row, &problem).await,
845 }
846 }
847 };
848 let total = combinations.len();
849 let raw = job.raw.as_object().cloned().unwrap_or_default();
Fast pages, required checks on the branch, self-hosted runners, honest incidents850 // Whether pull requests from forks may use self-hosted runners here,
851 // asked once, and only for a run that is not trusted.
852 let mut forks_allowed: Option<bool> = None;
Merge branch 'worktree-agent-a3abfcce648e87dca'853 // Each concurrency group its jobs join, and whether it cancels.
854 let mut groups: Vec<(String, bool)> = Vec::new();
GitHub Actions on g1t, part two: running workflows855 let mut statements = Vec::new();
856 for (index, combination) in combinations.iter().enumerate() {
857 let mut contexts = contexts.clone();
858 contexts.insert("matrix".into(), Value::Object(combination.clone()));
859 contexts.insert(
860 "strategy".into(),
861 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
862 );
863 let scope = Scope {
864 contexts: &contexts,
865 status: Status::Success,
866 hash_files: None,
867 };
868 let base_name = job.name.clone().unwrap_or(job.id.clone());
Actions: reusable workflows in the repository869 // A called workflow's job is shown under the job that called it.
870 let base_name = match &parent {
871 Some(parent) => format!("{} / {base_name}", parent.replace('/', " / ")),
872 None => base_name,
873 };
GitHub Actions on g1t, part two: running workflows874 let name = if expr::has_expression(&base_name) {
875 expr::interpolate(&base_name, &scope).unwrap_or(base_name)
876 } else if job.matrix.is_some() {
877 matrix::job_name(&base_name, combination)
878 } else {
879 base_name
880 };
881 let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null);
882 let labels = match &runs_on {
883 Value::String(label) => vec![label.clone()],
884 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
885 Value::Object(spec) => spec.get("labels").map(|l| match l {
886 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
887 other => vec![expr::to_text(other)],
888 }).unwrap_or_default(),
889 _ => Vec::new(),
890 };
Fast pages, required checks on the branch, self-hosted runners, honest incidents891 // `runs-on: self-hosted` (or a group): the workspace's own
892 // machines, which may run any OS. Otherwise g1t's Linux sandboxes.
893 let group = match &runs_on {
894 Value::Object(spec) => spec.get("group").map(expr::to_text),
895 _ => None,
896 };
897 let wanted = Wanted::of(&labels, group.as_deref());
898 let mut reason = if wanted.self_hosted {
899 None
900 } else {
901 labels
902 .iter()
903 .find(|label| {
904 let lower = label.to_ascii_lowercase();
905 lower.contains("windows") || lower.contains("macos")
906 })
907 .map(|label| {
908 let os = if label.to_ascii_lowercase().contains("windows") { "windows" } else { "macos" };
909 format!("`runs-on: {label}`: g1t's own runners are Linux. A self-hosted runner can run it: `runs-on: [self-hosted, {os}]`.")
910 })
911 };
912 // A pull request from a fork runs code anyone could write: never
913 // on the workspace's machines unless it said they may.
914 if wanted.self_hosted && run.trusted == 0 {
915 let allowed = match forks_allowed {
916 Some(allowed) => allowed,
917 None => {
918 let allowed = self.effective_runner_settings(&row.namespace, Some(&run.repo_id)).await?.fork_pull_requests;
919 forks_allowed = Some(allowed);
920 allowed
921 }
922 };
923 if !allowed {
924 reason = Some(
925 "Pull requests from forks do not run on self-hosted runners here. An admin can allow it under Settings, Runners.".to_owned(),
926 );
927 }
928 }
929 let max_minutes = if wanted.self_hosted { SELF_HOSTED_MAX_TIMEOUT_MINUTES } else { MAX_TIMEOUT_MINUTES };
GitHub Actions on g1t, part two: running workflows930 let timeout = raw
931 .get("timeout-minutes")
932 .and_then(|value| expr::interpolate_value(value, &scope).ok())
933 .and_then(|value| value.as_f64().or_else(|| expr::to_text(&value).parse().ok()))
Fast pages, required checks on the branch, self-hosted runners, honest incidents934 .map_or(MAX_TIMEOUT_MINUTES, |minutes| (minutes.ceil() as u32).clamp(1, max_minutes));
GitHub Actions on g1t, part two: running workflows935 let continue_on_error = raw
936 .get("continue-on-error")
937 .and_then(|value| expr::interpolate_value(value, &scope).ok())
938 .is_some_and(|value| expr::truthy(&value));
Merge branch 'worktree-agent-a3abfcce648e87dca'939 let (mut status, mut conclusion, mut finished) = match &reason {
GitHub Actions on g1t, part two: running workflows940 Some(_) => ("completed", Some("failure"), Some(now())),
941 None => ("queued", None, None),
942 };
Fast pages, required checks on the branch, self-hosted runners, honest incidents943 // A self-hosted job waits, saying for what, until a runner takes it.
944 let (labels_json, queued_at) = if wanted.self_hosted && reason.is_none() {
945 reason = Some(waiting_reason(&wanted));
946 (Some(serde_json::to_string(&wanted.stored())?), Some(now()))
947 } else {
948 (None, None)
949 };
Merge branch 'worktree-agent-a3abfcce648e87dca'950 // The environment it names, an expression read now, and what
951 // that environment's protection rules say: it starts, waits as
952 // `pending` (protection.rs), or may not deploy there.
953 let environment = environment_of(&job.raw, &contexts).map(|env| env.name);
954 if status == "queued"
955 && let Some(name) = &environment
956 {
957 match self.gate(run, name).await? {
958 Gate::Open => {}
959 Gate::Held(why) => {
960 status = "pending";
961 reason = Some(why);
962 }
963 Gate::Refused(why) => {
964 (status, conclusion, finished) = ("completed", Some("failure"), Some(now()));
965 reason = Some(why);
966 }
967 }
968 }
969 // Its own concurrency group.
970 let job_group = job
971 .concurrency
972 .as_ref()
973 .and_then(|c| expr::interpolate(&c.group, &scope).ok())
974 .map(|group| group.trim().to_owned())
975 .filter(|group| !group.is_empty());
976 let job_cancel = job
977 .concurrency
978 .as_ref()
979 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
980 .is_some_and(|value| expr::truthy(&value));
981 if status != "completed"
982 && let Some(group) = &job_group
983 && !groups.iter().any(|(known, _)| known == group)
984 {
985 groups.push((group.clone(), job_cancel));
986 }
GitHub Actions on g1t, part two: running workflows987 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
988 name.into(),
989 serde_json::to_string(combination)?.into(),
990 status.into(),
991 optional(conclusion),
992 optional(reason.as_deref()),
993 timeout.into(),
994 u32::from(continue_on_error).into(),
995 job.max_parallel.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
996 optional(finished.as_deref()),
Fast pages, required checks on the branch, self-hosted runners, honest incidents997 optional(labels_json.as_deref()),
998 optional(queued_at.as_deref()),
Merge branch 'worktree-agent-a3abfcce648e87dca'999 optional(environment.as_deref()),
1000 optional(job_group.as_deref()),
1001 u32::from(job_cancel).into(),
GitHub Actions on g1t, part two: running workflows1002 ];
1003 if index == 0 {
1004 let mut bound = values;
1005 bound.push(row.id.as_str().into());
1006 statements.push(
1007 self.db
1008 .prepare(
1009 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, timeout_minutes = ?,
Merge branch 'worktree-agent-a3abfcce648e87dca'1010 continue_on_error = ?, max_parallel = ?, finished_at = ?, labels = ?, queued_at = ?, environment = ?,
1011 concurrency_group = ?, cancel_in_progress = ? WHERE id = ?",
GitHub Actions on g1t, part two: running workflows1012 )
1013 .bind(&bound)?,
1014 );
1015 } else {
1016 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
1017 new_id("job", now_ms()).into(),
1018 row.run_id.as_str().into(),
1019 row.repo_id.as_str().into(),
1020 row.namespace.as_str().into(),
1021 row.key.as_str().into(),
1022 (index as u32).into(),
1023 row.needs.as_str().into(),
Actions: reusable workflows in the repository1024 optional(row.call.as_deref()),
GitHub Actions on g1t, part two: running workflows1025 ];
1026 bound.extend(values);
1027 statements.push(
1028 self.db
1029 .prepare(
Actions: reusable workflows in the repository1030 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, call, name, matrix, status, conclusion, reason,
Merge branch 'worktree-agent-a3abfcce648e87dca'1031 timeout_minutes, continue_on_error, max_parallel, finished_at, labels, queued_at, environment,
1032 concurrency_group, cancel_in_progress)
1033 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
GitHub Actions on g1t, part two: running workflows1034 )
1035 .bind(&bound)?,
1036 );
1037 }
1038 }
1039 self.db.batch(statements).await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1040 self.replace_in_groups(run, &row.key, &groups).await
1041 }
1042
1043 /// A job queued in its own concurrency group: with `cancel-in-progress`
1044 /// it cancels the group's other jobs; otherwise it replaces any that
1045 /// are queued and not started, and waits for the one running
1046 /// (`start_queued` starts one of a group at a time). As on GitHub.
1047 async fn replace_in_groups(&self, run: &RunRow, key: &str, groups: &[(String, bool)]) -> Result<()> {
1048 if groups.is_empty() {
1049 return Ok(());
1050 }
1051 #[derive(Deserialize)]
1052 struct Id {
1053 id: String,
1054 }
1055 let mine: Vec<String> = self
1056 .db
1057 .prepare("SELECT id FROM jobs WHERE run_id = ? AND key = ?")
1058 .bind(&[run.id.as_str().into(), key.into()])?
1059 .all()
1060 .await?
1061 .results::<Id>()?
1062 .into_iter()
1063 .map(|row| row.id)
1064 .collect();
1065 let mut moved: Vec<String> = Vec::new();
1066 for (group, cancel) in groups {
1067 let others = self
1068 .db
1069 .prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('queued', 'pending', 'in_progress')")
1070 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1071 .all()
1072 .await?
1073 .results::<JobRow>()?;
1074 for other in others.iter().filter(|other| !mine.contains(&other.id)) {
1075 if *cancel || other.status == "queued" {
1076 self.stop_job(other, "A newer job in the same concurrency group replaced it.").await?;
1077 if !moved.contains(&other.run_id) {
1078 moved.push(other.run_id.clone());
1079 }
1080 }
1081 }
1082 }
1083 for other_run in moved.iter().filter(|id| **id != run.id) {
1084 Box::pin(self.advance(other_run)).await?;
1085 }
GitHub Actions on g1t, part two: running workflows1086 Ok(())
1087 }
1088
Actions: reusable workflows in the repository1089 /// A job that calls a reusable workflow in the repository: that
1090 /// workflow's jobs join the run under it, with the inputs it passes.
1091 async fn call_workflow(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, uses: &str, scope: &Scope<'_>) -> Result<()> {
1092 let Some(local) = uses.strip_prefix("./") else {
1093 return self
1094 .fail_job(row, "Reusable workflows from other repositories are not called on g1t yet; ones in this repository (`./.g1t/workflows/…`) are.")
1095 .await;
1096 };
1097 let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;
1098 if depth > MAX_CALL_DEPTH {
1099 return self.fail_job(row, &format!("Reusable workflows call each other more than {MAX_CALL_DEPTH} deep.")).await;
1100 }
1101 let local = local.split('@').next().unwrap_or(local).to_owned();
1102 let path = repo_path(&run.repo);
1103 let Some(ws) = self.workspace_actor(&path.namespace).await? else {
1104 return self.fail_job(row, "The workspace is gone.").await;
1105 };
1106 // A repository moved from GitHub keeps saying `.github/…`.
1107 let mut found = self.read_file(&path, &ws, &run.sha, &local).await?.map(|text| (local.clone(), text));
1108 if found.is_none()
1109 && let Some(rest) = local.strip_prefix(".github/")
1110 {
1111 let moved = format!(".g1t/{rest}");
1112 found = self.read_file(&path, &ws, &run.sha, &moved).await?.map(|text| (moved, text));
1113 }
1114 let Some((file, source)) = found else {
1115 return self.fail_job(row, &format!("`{uses}` is not in the repository at this commit.")).await;
1116 };
1117 let called = match workflow::parse(&source) {
1118 Ok(called) => called,
1119 Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,
1120 };
1121 let Some(trigger) = called.trigger("workflow_call") else {
1122 return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;
1123 };
1124 // Inputs: what the caller passes, else the called workflow's defaults.
1125 let given = match job.raw.get("with") {
1126 Some(with) => match expr::interpolate_value(with, scope) {
1127 Ok(Value::Object(given)) => given,
1128 Ok(_) => Map::new(),
1129 Err(problem) => return self.fail_job(row, &format!("Its `with` does not read: {problem}")).await,
1130 },
1131 None => Map::new(),
1132 };
1133 let mut inputs = Map::new();
1134 for (name, spec) in &trigger.inputs {
1135 let value = given.get(name).cloned().or_else(|| spec.get("default").cloned()).unwrap_or(Value::Null);
1136 if value.is_null() && spec.get("required").and_then(Value::as_bool) == Some(true) {
1137 return self.fail_job(row, &format!("`{file}` needs the input `{name}`.")).await;
1138 }
1139 inputs.insert(name.clone(), value);
1140 }
1141 for (name, value) in given {
1142 inputs.entry(name).or_insert(value);
1143 }
1144 let mut statements = Vec::new();
1145 for called_job in &called.jobs {
1146 let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{}/{n}", row.key)).collect();
1147 let call = json!({
1148 "role": "callee", "parent": row.key, "job": called_job.id, "path": file,
1149 "source": source, "inputs": inputs, "depth": depth,
1150 });
1151 statements.push(
1152 self.db
1153 .prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")
1154 .bind(&[
1155 new_id("job", now_ms()).into(),
1156 row.run_id.as_str().into(),
1157 row.repo_id.as_str().into(),
1158 row.namespace.as_str().into(),
1159 format!("{}/{}", row.key, called_job.id).into(),
1160 format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),
1161 serde_json::to_string(&needs)?.into(),
1162 serde_json::to_string(&call)?.into(),
1163 ])?,
1164 );
1165 }
1166 statements.push(
1167 self.db
1168 .prepare("UPDATE jobs SET status = 'calling', call = ?, reason = ?, started_at = ? WHERE id = ?")
1169 .bind(&[
1170 serde_json::to_string(&json!({ "role": "caller", "path": file, "source": source }))?.into(),
1171 format!("Calls `{file}`.").into(),
1172 now().into(),
1173 row.id.as_str().into(),
1174 ])?,
1175 );
1176 self.db.batch(statements).await?;
1177 Ok(())
1178 }
1179
1180 /// A job that called a workflow, finished with its jobs: their result,
1181 /// and the outputs the workflow declares.
1182 async fn finish_call(&self, row: &JobRow, children: &[&JobRow]) -> Result<()> {
1183 let call = row.call().unwrap_or_default();
1184 let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());
1185 let mut jobs_context = Map::new();
1186 let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();
1187 for child in children {
1188 by_key.entry(child.key.clone()).or_default().push(child);
1189 }
1190 for (key, rows) in &by_key {
1191 let mut outputs = Map::new();
1192 for child in rows {
1193 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {
1194 outputs.extend(more);
1195 }
1196 }
1197 let id = key.rsplit('/').next().unwrap_or(key);
1198 jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));
1199 }
1200 let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));
1201 let mut contexts = Map::new();
1202 contexts.insert("jobs".into(), Value::Object(jobs_context));
1203 contexts.insert("inputs".into(), inputs);
1204 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1205 let mut outputs = Map::new();
1206 if let Some(Value::Object(declared)) = called.as_ref().map(|w| {
1207 let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);
1208 on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)
1209 }) {
1210 for (name, spec) in declared {
1211 if let Some(value) = spec.get("value") {
1212 let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);
1213 outputs.insert(name, Value::String(expr::to_text(&value)));
1214 }
1215 }
1216 }
1217 let conclusion = key_result(children);
1218 self.db
1219 .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling'")
1220 .bind(&[conclusion.into(), serde_json::to_string(&outputs)?.into(), now().into(), row.id.as_str().into()])?
1221 .run()
1222 .await?;
1223 Ok(())
1224 }
1225
1226 /// A file's text at a commit, if it is there.
1227 async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {
1228 let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(
1229 &self.repos,
1230 "blob",
1231 &g1t_contracts::repos::BlobArgs {
1232 path: path.clone(),
1233 viewer: Some(ws.clone()),
1234 git_ref: sha.to_owned(),
1235 file_path: file.to_owned(),
1236 },
1237 )
1238 .await?;
1239 Ok(match blob {
1240 Outcome::Ok(view) => view.text,
1241 Outcome::Fail(_) => None,
1242 })
1243 }
1244
GitHub Actions on g1t, part two: running workflows1245 async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {
1246 self.db
1247 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")
1248 .bind(&[optional(reason), now().into(), row.id.as_str().into()])?
1249 .run()
1250 .await?;
1251 Ok(())
1252 }
1253
1254 async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {
1255 self.db
1256 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")
1257 .bind(&[reason.into(), now().into(), row.id.as_str().into()])?
1258 .run()
1259 .await?;
1260 Ok(())
1261 }
1262
1263 /// Starts queued jobs, oldest first, while their workspace has room.
1264 pub async fn start_queued(&self) -> Result<()> {
1265 let queued = self
1266 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1267 // Self-hosted jobs are taken by their runners (runners.rs).
1268 .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NULL ORDER BY rowid LIMIT 50")
GitHub Actions on g1t, part two: running workflows1269 .all()
1270 .await?
1271 .results::<JobRow>()?;
1272 let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
1273 for job in queued {
1274 let in_workspace = match running.get(&job.namespace) {
1275 Some(n) => *n,
1276 None => {
1277 let n = self
1278 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1279 // Only g1t's own sandboxes count against the workspace's room.
1280 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND namespace = ? AND runner_id IS NULL")
GitHub Actions on g1t, part two: running workflows1281 .bind(&[job.namespace.as_str().into()])?
1282 .first::<Count>(None)
1283 .await?
1284 .map_or(0, |count| count.n);
1285 running.insert(job.namespace.clone(), n);
1286 n
1287 }
1288 };
1289 if in_workspace >= RUNNING_PER_WORKSPACE {
1290 continue;
1291 }
1292 if let Some(max) = job.max_parallel {
1293 let siblings = self
1294 .db
1295 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
1296 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1297 .first::<Count>(None)
1298 .await?
1299 .map_or(0, |count| count.n);
1300 if siblings >= max {
1301 continue;
1302 }
1303 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1304 // One job of a concurrency group runs at a time.
1305 if let Some(group) = &job.concurrency_group {
1306 let running = self
1307 .db
1308 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'in_progress' AND id != ?")
1309 .bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])?
1310 .first::<Count>(None)
1311 .await?
1312 .map_or(0, |count| count.n);
1313 if running > 0 {
1314 continue;
1315 }
1316 }
GitHub Actions on g1t, part two: running workflows1317 let token = random_hex(24);
1318 let at = now();
1319 let claimed = self
1320 .db
1321 .prepare(
1322 "UPDATE jobs SET status = 'in_progress', token_hash = ?, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
1323 )
1324 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1325 .first::<Value>(None)
1326 .await?;
1327 if claimed.is_none() {
1328 continue;
1329 }
1330 running.insert(job.namespace.clone(), in_workspace + 1);
1331 self.db
1332 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
1333 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?
1334 .run()
1335 .await?;
1336 let run = self.run_row(&job.run_id).await?;
1337 let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {
1338 namespace: job.namespace.clone(),
1339 name: String::new(),
1340 });
Fast pages, required checks on the branch, self-hosted runners, honest incidents1341 // Its environment and the machine it asked for, for the runner.
1342 let details = run.as_ref().map(|run| start_details(run, &job)).unwrap_or_default();
GitHub Actions on g1t, part two: running workflows1343 let started: Outcome<Value> = g1t_kit::call(
1344 &self.runner,
1345 "start_actions_job",
1346 &StartJobArgs {
1347 job: job.id.clone(),
1348 token,
1349 repo,
1350 timeout_minutes: job.timeout_minutes,
Fast pages, required checks on the branch, self-hosted runners, honest incidents1351 workflow: run.as_ref().map(|run| run.path.clone()),
1352 environment: details.environment,
1353 trusted: run.as_ref().is_some_and(|run| run.trusted != 0),
1354 instance: details.instance,
GitHub Actions on g1t, part two: running workflows1355 },
1356 )
1357 .await
1358 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
1359 if let Outcome::Fail(refused) = started {
1360 Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971361 } else {
1362 self.job_started(&job.id).await?;
GitHub Actions on g1t, part two: running workflows1363 }
1364 }
1365 Ok(())
1366 }
1367
1368 /// Finishes a job and moves its run along.
1369 pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {
1370 let finished = self
1371 .db
1372 .prepare(
1373 "UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
1374 finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
1375 )
1376 .bind(&[
1377 conclusion.into(),
1378 optional(reason),
1379 outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1380 now().into(),
1381 job_id.into(),
1382 ])?
1383 .first::<JobRow>(None)
1384 .await?;
1385 let Some(job) = finished else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1386 // Its G1T_TOKEN stops working with it.
1387 self.revoke_job_tokens(&job.id).await;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971388 // A job that deploys failed: so did its run's deployment, now.
1389 if conclusion == "failure" && job.continue_on_error == 0 && job.started_at.is_some()
1390 && let Some(run) = self.run_row(&job.run_id).await?
1391 && let Some(env) = deploys_to(&run, &job)
1392 {
1393 self.report_deployment(&run, &env, "failure", false).await;
1394 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1395 // A self-hosted runner's job: the runner is free again, and its time
1396 // is recorded, at nothing.
1397 if job.runner_id.is_some() {
1398 self.released(&job).await?;
1399 }
GitHub Actions on g1t, part two: running workflows1400 // Steps still marked as going are not going any more.
1401 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1402 let mut touched = false;
1403 for step in steps.iter_mut() {
1404 if step["status"] != "completed" {
1405 let was_running = step["status"] == "in_progress";
1406 step["status"] = json!("completed");
1407 step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });
1408 touched = true;
1409 }
1410 }
1411 if touched {
1412 self.db
1413 .prepare("UPDATE jobs SET steps = ? WHERE id = ?")
1414 .bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?
1415 .run()
1416 .await?;
1417 }
1418 // fail-fast: one failed combination stops the rest of its matrix.
1419 if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {
1420 let run = self.run_row(&job.run_id).await?;
1421 let fail_fast = run
1422 .as_ref()
1423 .and_then(|run| workflow::parse(&run.source).ok())
1424 .and_then(|workflow| workflow.jobs.into_iter().find(|j| j.id == job.key))
1425 .is_none_or(|j| j.fail_fast);
1426 if fail_fast {
1427 let siblings = self
1428 .db
1429 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1430 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1431 .all()
1432 .await?
1433 .results::<JobRow>()?;
1434 for sibling in siblings {
1435 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1436 }
1437 }
1438 }
1439 Box::pin(self.advance(&job.run_id)).await
1440 }
1441
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1442 /// Cancels a job. One that is running is told to stop (the answer to
1443 /// its next report), and runs its `if: always()` and `cancelled()`
1444 /// steps and its post steps before it ends `cancelled`; one that has
1445 /// not started is cancelled at once.
GitHub Actions on g1t, part two: running workflows1446 async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1447 if job.status == "in_progress" && job.started_at.is_some() {
1448 self.db
1449 .prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")
1450 .bind(&[now().into(), reason.into(), job.id.as_str().into()])?
1451 .run()
1452 .await?;
1453 return Ok(());
1454 }
1455 self.hard_stop(job, reason).await
1456 }
1457
1458 /// Cancels a job outright, stopping its sandbox if it has one.
1459 async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {
Fast pages, required checks on the branch, self-hosted runners, honest incidents1460 // A self-hosted runner hears it was cancelled on its next poll.
1461 if job.status == "in_progress" && job.runner_id.is_none() {
GitHub Actions on g1t, part two: running workflows1462 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
1463 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1464 let stopped = self
1465 .db
1466 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")
GitHub Actions on g1t, part two: running workflows1467 .bind(&[reason.into(), now().into(), job.id.as_str().into()])?
Fast pages, required checks on the branch, self-hosted runners, honest incidents1468 .first::<JobRow>(None)
GitHub Actions on g1t, part two: running workflows1469 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1470 if stopped.as_ref().is_some_and(|row| row.started_at.is_some()) {
1471 self.revoke_job_tokens(&job.id).await;
1472 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1473 if let Some(stopped) = stopped.filter(|row| row.runner_id.is_some()) {
1474 self.released(&stopped).await?;
1475 }
GitHub Actions on g1t, part two: running workflows1476 Ok(())
1477 }
1478
Merge branch 'worktree-agent-a3abfcce648e87dca'1479 /// Ends a job's tokens at once. A failure is logged: the token expires
1480 /// on its own soon after the job's time limit.
1481 async fn revoke_job_tokens(&self, job_id: &str) {
1482 let revoked: Result<bool> =
1483 g1t_kit::call(&self.identity, "revoke_job_tokens", &RevokeJobTokensArgs { job_id: job_id.to_owned() }).await;
1484 if let Err(error) = revoked {
1485 worker::console_error!("actions: the tokens of job {job_id} were not revoked: {error}");
1486 }
1487 }
1488
GitHub Actions on g1t, part two: running workflows1489 /// Finishes the run when every job has.
1490 async fn finish_if_done(&self, run_id: &str) -> Result<()> {
1491 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1492 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
GitHub Actions on g1t, part two: running workflows1493 return Ok(());
1494 }
1495 let jobs = self.job_rows(run_id).await?;
1496 if !jobs.iter().all(|job| job.status == "completed") {
1497 return Ok(());
1498 }
1499 self.finish_run(&run, None).await
1500 }
1501
1502 async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {
1503 let jobs = self.job_rows(&run.id).await?;
1504 let rows: Vec<&JobRow> = jobs.iter().collect();
1505 let conclusion = if error.is_some() {
1506 "failure"
1507 } else if run.conclusion.as_deref() == Some("cancelled") {
1508 "cancelled"
1509 } else if rows.is_empty() {
1510 "skipped"
1511 } else {
1512 key_result(&rows)
1513 };
1514 let done = self
1515 .db
1516 .prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")
1517 .bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?
1518 .first::<Value>(None)
1519 .await?;
1520 if done.is_none() {
1521 return Ok(());
1522 }
1523 self.report_status(run, conclusion).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971524 self.settle_deployments(run, &jobs, conclusion == "cancelled").await;
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 241525 let published: Result<()> = g1t_kit::call(
1526 &self.events,
1527 "publish",
1528 &g1t_contracts::events::Publish {
1529 events: vec![g1t_contracts::events::NewEvent {
1530 kind: "workflow.completed",
1531 source: "actions",
1532 repo_id: Some(run.repo_id.clone()),
1533 actor: run.actor_id.clone(),
1534 data: g1t_contracts::events::WorkflowEvent {
1535 run_id: run.id.clone(),
1536 repo_id: run.repo_id.clone(),
1537 workflow: run.name.clone(),
1538 path: run.path.clone(),
1539 number: run.number,
1540 event: run.event.clone(),
1541 conclusion: conclusion.to_owned(),
1542 git_ref: run.git_ref.clone(),
1543 sha: run.sha.clone(),
1544 pull: run.pull,
1545 },
1546 }],
1547 },
1548 )
1549 .await;
1550 if let Err(error) = published {
1551 worker::console_error!("actions: could not publish workflow.completed: {error}");
1552 }
GitHub Actions on g1t, part two: running workflows1553 // The next run waiting in its concurrency group.
1554 if let Some(group) = &run.concurrency_group {
1555 let next = self
1556 .db
1557 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")
1558 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1559 .first::<RunRow>(None)
1560 .await?;
1561 if let Some(next) = next {
1562 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;
1563 Box::pin(self.advance(&next.id)).await?;
1564 }
1565 }
1566 Ok(())
1567 }
1568
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971569 /// Tells the deployments service how a run's deployment to `env` stands:
1570 /// made when its first job naming the environment starts, failed when one
1571 /// of them fails, and settled (`last`) when the run finishes. Never fails
1572 /// the run: a deployment that could not be recorded is logged.
1573 async fn report_deployment(&self, run: &RunRow, env: &JobEnvironment, state: &str, last: bool) {
1574 let path = repo_path(&run.repo);
1575 let reported: Result<Value> = g1t_kit::call(
1576 &self.deployments,
1577 "actions_deployment",
1578 &json!({
1579 "repoId": run.repo_id,
1580 "repo": { "namespace": path.namespace, "name": path.name },
1581 "runId": run.id,
1582 "attempt": run.attempt,
1583 "runUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1584 "environment": env.name,
1585 "url": env.url,
1586 "ref": run.git_ref,
1587 "sha": run.sha,
1588 "state": state,
1589 "final": last,
1590 "creator": run.actor,
1591 "workflow": run.name,
1592 }),
1593 )
1594 .await;
1595 if let Err(error) = reported {
1596 worker::console_error!("actions: deployment to {} not recorded for run {}: {error}", env.name, run.id);
1597 }
1598 }
1599
1600 /// A job that deploys has started: its run's deployment to the
1601 /// environment is under way.
1602 pub(crate) async fn job_started(&self, job_id: &str) -> Result<()> {
1603 let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[job_id.into()])?.first::<JobRow>(None).await? else {
1604 return Ok(());
1605 };
1606 let Some(run) = self.run_row(&job.run_id).await? else { return Ok(()) };
1607 if let Some(env) = deploys_to(&run, &job) {
1608 self.report_deployment(&run, &env, "in_progress", false).await;
1609 }
1610 Ok(())
1611 }
1612
1613 /// The run is over: each environment its jobs deployed to takes the
1614 /// outcome of those jobs (`deployment_outcome`).
1615 async fn settle_deployments(&self, run: &RunRow, jobs: &[JobRow], cancelled: bool) {
1616 let mut seen: Vec<(JobEnvironment, Vec<Option<String>>)> = Vec::new();
1617 for job in jobs {
1618 let Some(env) = deploys_to(run, job) else { continue };
1619 let conclusion = if cancelled && job.conclusion.as_deref() != Some("skipped") && job.started_at.is_some() {
1620 Some("cancelled".to_owned())
1621 } else if job.started_at.is_none() {
1622 Some("skipped".to_owned())
1623 } else {
1624 job.conclusion.clone()
1625 };
1626 match seen.iter_mut().find(|(known, _)| known.name.eq_ignore_ascii_case(&env.name)) {
1627 Some((known, conclusions)) => {
1628 if known.url.is_none() {
1629 known.url = env.url.clone();
1630 }
1631 conclusions.push(conclusion);
1632 }
1633 None => seen.push((env, vec![conclusion])),
1634 }
1635 }
1636 for (env, conclusions) in seen {
1637 if let Some(state) = deployment_outcome(&conclusions) {
1638 self.report_deployment(run, &env, state, true).await;
1639 }
1640 }
1641 }
1642
GitHub Actions on g1t, part two: running workflows1643 /// Tells the pull request (or commit) how the run went, as a status.
1644 async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {
1645 let state = match conclusion {
1646 "success" | "skipped" => "success",
1647 "cancelled" => "error",
1648 _ => "failure",
1649 };
1650 let _: Result<Value> = g1t_kit::call(
1651 &self.work,
1652 "set_commit_status",
1653 &json!({
1654 "repoId": run.repo_id,
1655 "sha": run.sha,
1656 "context": format!("{} / {}", run.name, run.event),
1657 "state": state,
1658 "description": format!("{} {}", run.name, match conclusion {
1659 "success" => "passed",
1660 "skipped" => "was skipped",
1661 "cancelled" => "was cancelled",
1662 _ => "failed",
1663 }),
1664 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1665 "source": "actions",
GitHub Actions on g1t, part two: running workflows1666 }),
1667 )
1668 .await;
1669 Ok(())
1670 }
1671
1672 /// Tells the pull request a run has started on its head.
1673 pub async fn report_pending(&self, run: &RunRow) -> Result<()> {
1674 let _: Result<Value> = g1t_kit::call(
1675 &self.work,
1676 "set_commit_status",
1677 &json!({
1678 "repoId": run.repo_id,
1679 "sha": run.sha,
1680 "context": format!("{} / {}", run.name, run.event),
1681 "state": "pending",
Merge branch 'worktree-agent-a3abfcce648e87dca'1682 "description": if run.status == "action_required" {
1683 format!("{} is waiting for approval", run.name)
1684 } else {
1685 format!("{} is running", run.name)
1686 },
GitHub Actions on g1t, part two: running workflows1687 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1688 "source": "actions",
GitHub Actions on g1t, part two: running workflows1689 }),
1690 )
1691 .await;
1692 Ok(())
1693 }
1694
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1695 /// Cancels a run: its waiting and queued jobs at once, and its running
1696 /// ones gracefully (`stop_job`), or outright when `force`.
1697 pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {
GitHub Actions on g1t, part two: running workflows1698 self.db
1699 .prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
1700 .bind(&[run.id.as_str().into()])?
1701 .run()
1702 .await?;
1703 for job in self.job_rows(&run.id).await?.iter().filter(|job| job.status != "completed") {
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1704 if force {
1705 self.hard_stop(job, reason).await?;
1706 } else {
1707 self.stop_job(job, reason).await?;
1708 }
GitHub Actions on g1t, part two: running workflows1709 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1710 if run.status == "pending" || run.status == "action_required" {
GitHub Actions on g1t, part two: running workflows1711 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
1712 }
1713 self.advance(&run.id).await
1714 }
1715
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1716 /// Cancels every run of the repository that has not finished, for
1717 /// `repo.deleted` and `repo.archived`.
1718 pub async fn stop_runs(&self, repo_id: &str) -> Result<()> {
1719 let runs = self
1720 .db
1721 .prepare("SELECT * FROM runs WHERE repo_id = ? AND status != 'completed'")
1722 .bind(&[repo_id.into()])?
1723 .all()
1724 .await?
1725 .results::<RunRow>()?;
1726 for run in runs {
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1727 self.cancel_run(&run, "The repository was archived or deleted.", true).await?;
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1728 }
1729 Ok(())
1730 }
1731
GitHub Actions on g1t, part two: running workflows1732 pub async fn cancel(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1733 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
GitHub Actions on g1t, part two: running workflows1734 return Ok(Outcome::Fail(refused));
1735 }
1736 let run = check!(self.run_in(&a.repo, &a.id).await?);
1737 if run.status == "completed" {
1738 return Ok(fail(FailureCode::Conflict, "The run has already finished."));
1739 }
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1740 // Cancelling a run that is already cancelling stops its jobs
1741 // outright, without waiting for their cleanup steps.
1742 let force = a.force || run.conclusion.as_deref() == Some("cancelled");
1743 let reason = if force {
1744 format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)
1745 } else {
1746 format!("{} cancelled the run.", a.actor.username)
1747 };
1748 self.cancel_run(&run, &reason, force).await?;
GitHub Actions on g1t, part two: running workflows1749 self.run_summary(&run.id).await
1750 }
1751
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1752 /// Runs again, as a new attempt: every job, with `failed_only` those
1753 /// that did not succeed, or with `job` that one; each with the jobs that
1754 /// need them. The attempt that ends is kept, its jobs and their logs and
1755 /// summaries with it (`job_attempts`, `run_attempts`).
GitHub Actions on g1t, part two: running workflows1756 pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1757 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
GitHub Actions on g1t, part two: running workflows1758 return Ok(Outcome::Fail(refused));
1759 }
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1760 // A job alone (`POST …/jobs/{job}/rerun`) names its run.
1761 let run_id = match (&a.job, a.id.is_empty()) {
1762 (Some(job), true) => {
1763 #[derive(Deserialize)]
1764 struct Of {
1765 run_id: String,
1766 }
1767 let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;
1768 match of {
1769 Some(of) => of.run_id,
1770 None => return Ok(fail(FailureCode::NotFound, "No such job.")),
1771 }
1772 }
1773 _ => a.id.clone(),
1774 };
1775 let run = check!(self.run_in(&a.repo, &run_id).await?);
GitHub Actions on g1t, part two: running workflows1776 if run.status != "completed" {
1777 return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
1778 }
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1779 // Nothing starts again on an archived repository.
1780 match self.visible_repo(&a.repo, &Some(a.actor.clone())).await? {
1781 Some(repo) if repo.archived() => {
1782 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
1783 }
1784 Some(_) => {}
1785 None => return Ok(fail(FailureCode::NotFound, "There is no such repository.")),
1786 }
GitHub Actions on g1t, part two: running workflows1787 if run.error.is_some() {
1788 return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));
1789 }
1790 let jobs = self.job_rows(&run.id).await?;
1791 let workflow = workflow::parse(&run.source).ok();
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1792 let target = match &a.job {
1793 Some(id) => match jobs.iter().find(|job| &job.id == id) {
1794 Some(job) => Some(top_key(&job.key).to_owned()),
1795 None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),
1796 },
1797 None => None,
1798 };
1799 let which = match (&target, a.failed_only) {
1800 (Some(key), _) => Rerun::Job(key),
1801 (None, true) => Rerun::Failed,
1802 (None, false) => Rerun::All,
1803 };
1804 let order = workflow.as_ref().map(|w| w.job_order()).unwrap_or_default();
1805 let again = rerun_keys(
1806 &order,
1807 |key| jobs.iter().find(|j| j.key == key).map(JobRow::needs).unwrap_or_default(),
1808 |key| jobs.iter().filter(|j| j.key == key).all(|j| j.conclusion.as_deref() == Some("success")),
1809 which,
1810 );
GitHub Actions on g1t, part two: running workflows1811 if again.is_empty() {
1812 return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
1813 }
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1814 // The jobs that run again, a called workflow's with the job calling it.
1815 let rerun_ids: Vec<&str> = jobs.iter().filter(|job| again.iter().any(|key| key == top_key(&job.key))).map(|job| job.id.as_str()).collect();
1816 let ids = serde_json::to_string(&rerun_ids)?;
1817 let ended = run.attempt.to_string();
1818 let ended = ended.as_str();
1819 let mut statements = vec![
1820 // The attempt that ends, as it ended.
1821 self.db
1822 .prepare(
1823 "INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)
1824 SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",
1825 )
1826 .bind(&[run.id.as_str().into()])?,
1827 // Earlier attempts that showed a job's logs from where it ran
1828 // then now find them where they move to.
1829 self.db
1830 .prepare("UPDATE job_attempts SET log_id = log_id || '.' || ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")
1831 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
1832 // Each of its jobs: one that runs again keeps its logs under
1833 // `{id}.{attempt}`; one left alone is still the live job's.
1834 self.db
1835 .prepare(
1836 "INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,
1837 steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)
1838 SELECT id || '.' || ?1, run_id, repo_id, ?1, id,
1839 CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id || '.' || ?1 ELSE id END,
1840 key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at
1841 FROM jobs WHERE run_id = ?2 ORDER BY rowid",
1842 )
1843 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
1844 self.db
1845 .prepare("UPDATE logs SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
1846 .bind(&[ended.into(), ids.as_str().into()])?,
1847 self.db
1848 .prepare("UPDATE job_summaries SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
1849 .bind(&[ended.into(), ids.as_str().into()])?,
1850 ];
GitHub Actions on g1t, part two: running workflows1851 for key in &again {
1852 statements.push(self.db.prepare("DELETE FROM jobs WHERE run_id = ? AND key = ? AND ordinal > 0").bind(&[run.id.as_str().into(), key.as_str().into()])?);
Actions: reusable workflows in the repository1853 // The jobs of a workflow it called are made again when it calls it again.
GitHub Actions on g1t, part two: running workflows1854 statements.push(
1855 self.db
Actions: reusable workflows in the repository1856 .prepare("DELETE FROM jobs WHERE run_id = ? AND key LIKE ?")
1857 .bind(&[run.id.as_str().into(), format!("{key}/%").into()])?,
1858 );
1859 statements.push(
1860 self.db
GitHub Actions on g1t, part two: running workflows1861 .prepare(
1862 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Fast pages, required checks on the branch, self-hosted runners, honest incidents1863 matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
Merge branch 'worktree-agent-a3abfcce648e87dca'1864 labels = NULL, queued_at = NULL, runner_id = NULL, runner_name = NULL, environment = NULL,
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1865 concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
GitHub Actions on g1t, part two: running workflows1866 )
1867 .bind(&[run.id.as_str().into(), key.as_str().into()])?,
1868 );
1869 }
1870 statements.push(
1871 self.db
Actions: keep every run attempt, re-run one job, graceful cancel, summaries1872 .prepare(
1873 "UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,
1874 triggering_actor = ?, debug = ? WHERE id = ?",
1875 )
1876 .bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,
GitHub Actions on g1t, part two: running workflows1877 );
1878 self.db.batch(statements).await?;
1879 if let Some(run) = self.run_row(&run.id).await? {
1880 self.report_pending(&run).await?;
1881 }
1882 self.advance(&run.id).await?;
1883 self.run_summary(&run.id).await
1884 }
1885
1886 pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {
1887 let row = self
1888 .db
1889 .prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")
1890 .bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?
1891 .first::<RunRow>(None)
1892 .await?;
1893 Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))
1894 }
1895
1896 // --- The sandbox's side -----------------------------------------------------
1897
Fast pages, required checks on the branch, self-hosted runners, honest incidents1898 pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
GitHub Actions on g1t, part two: running workflows1899 let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
1900 Ok(match job {
1901 Some(job) if job.status == "in_progress" && job.token_hash.as_deref().is_some_and(|hash| same(hash, &sha256_hex(&a.token))) => {
1902 Outcome::Ok(job)
1903 }
1904 _ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
1905 })
1906 }
1907
A repository has its own sidebar, as settings do1908 /// `job_auth`: which run and repository a running job's token is for,
1909 /// so the API can keep its artifacts and cache.
1910 pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1911 let job = check!(self.job_for_token(&a).await?);
1912 Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))
1913 }
1914
GitHub Actions on g1t, part two: running workflows1915 /// `job_spec`: everything the sandbox needs to run the job.
1916 pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1917 let job = check!(self.job_for_token(&a).await?);
1918 let Some(run) = self.run_row(&job.run_id).await? else {
1919 return Ok(fail(FailureCode::NotFound, "No such run."));
1920 };
Actions: reusable workflows in the repository1921 let Ok(caller) = workflow::parse(&run.source) else {
GitHub Actions on g1t, part two: running workflows1922 return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));
1923 };
Actions: reusable workflows in the repository1924 // A called workflow's job runs as that workflow defines it.
1925 let callee = job.callee();
1926 let (workflow, spec, call_inputs) = match callee {
1927 Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),
1928 None => match caller.jobs.iter().find(|j| j.id == job.key) {
1929 Some(spec) => (caller.clone(), spec.clone(), None),
1930 None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),
1931 },
GitHub Actions on g1t, part two: running workflows1932 };
Actions: reusable workflows in the repository1933 let spec = &spec;
GitHub Actions on g1t, part two: running workflows1934 let repo = repo_path(&run.repo);
1935 let trusted = run.trusted != 0;
Merge branch 'worktree-agent-a3abfcce648e87dca'1936 // The job's `environment:`, by name, as read when its needs were done
1937 // (an expression included), once the environment's protection rules
1938 // let it start: entries with a value for it give that value instead
1939 // of their default, as GitHub's environment secrets do.
1940 let environment: Option<String> = job.environment.clone();
1941 // What its token may do: its `permissions:` (a called workflow's
1942 // jobs no more than the job that calls it), else the repository's
1943 // default; read-only for a pull request from outside.
1944 // The repository's default, its workspace's taken in, and whether
1945 // its jobs may open and approve pull requests.
1946 let (default, pull_requests) = self.token_policy(&run.repo_id).await?;
1947 let mut permissions = spec.permissions(&workflow, default);
1948 if let Some(parent) = &job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned)) {
1949 let top = parent.split('/').next().unwrap_or(parent);
1950 if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) {
1951 permissions = permissions.capped_by(&caller_job.permissions(&caller, default));
GitHub Actions on g1t, part two: running workflows1952 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1953 }
1954 if !trusted {
1955 permissions = permissions.read_only();
1956 }
1957 // G1T_TOKEN, and GITHUB_TOKEN as its alias: a token of the
1958 // workspace's that reaches this repository only, with the scopes
1959 // its permissions give, until the job ends.
1960 let token = match self.workspace_actor(&repo.namespace).await? {
1961 Some(workspace) => {
1962 let created: CreatedAccessToken = g1t_kit::call(
1963 &self.identity,
1964 "create_job_token",
1965 &CreateJobTokenArgs {
1966 workspace,
1967 repo: repo.clone(),
1968 run_id: run.id.clone(),
1969 job_id: job.id.clone(),
1970 name: format!("G1T_TOKEN for {} run {}", run.repo, run.number),
1971 ttl_seconds: u64::from(job.timeout_minutes) * 60 + 600,
1972 scopes: permissions.scopes().into_iter().map(str::to_owned).collect(),
1973 pull_requests: pull_requests && trusted,
1974 },
1975 )
1976 .await?;
1977 created.token
1978 }
1979 None => String::new(),
GitHub Actions on g1t, part two: running workflows1980 };
Secrets and variables: one list, rows per environment, for workflows and deployments1981 // A run that is not trusted (a pull request from outside the
1982 // workspace) gets no secrets and an empty token.
1983 let mut secrets = if trusted {
1984 self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?
1985 } else {
1986 Map::new()
1987 };
1988 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
GitHub Actions on g1t, part two: running workflows1989 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
Merge branch 'worktree-agent-a3abfcce648e87dca'1990 // Each secret as it is, a line at a time, base64 and JSON-escaped.
1991 let mut masks: Vec<String> = g1t_actions::mask::all_variants(secrets.values().filter_map(|v| v.as_str()));
Secrets and variables: one list, rows per environment, for workflows and deployments1992 let vars = self.variables_for(&run.repo_id, &run.repo, environment.as_deref(), trusted).await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R21993 // The toolkit's runtime token (runtime.rs), for as long as the job
1994 // may run; the API puts it and the toolkit's addresses in the
1995 // job's variables.
1996 let runtime_token = crate::runtime::runtime_token(
1997 &job.id,
1998 &job.run_id,
1999 job.token_hash.as_deref().unwrap_or_default(),
2000 now_ms() / 1000,
2001 u64::from(job.timeout_minutes) * 60 + 600,
2002 );
2003 masks.push(runtime_token.clone());
2004 let retention_days = self.retention_setting(&run.repo_id).await?;
GitHub Actions on g1t, part two: running workflows2005
2006 let jobs = self.job_rows(&run.id).await?;
2007 let mut needs = Map::new();
Actions: reusable workflows in the repository2008 // In a called workflow, its jobs' keys sit under the job that called it.
2009 let parent = job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
GitHub Actions on g1t, part two: running workflows2010 for need in &spec.needs {
Actions: reusable workflows in the repository2011 let key = match &parent {
2012 Some(parent) => format!("{parent}/{need}"),
2013 None => need.clone(),
2014 };
2015 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
GitHub Actions on g1t, part two: running workflows2016 let mut outputs = Map::new();
2017 for row in &rows {
2018 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
2019 outputs.extend(more);
2020 }
2021 }
2022 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
2023 }
2024 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
2025 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
2026 let info = run.info();
2027 let mut github = info.context(&job.key, &token, run.action.as_deref());
2028 github["token"] = json!(token);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22029 github["retention_days"] = json!(retention_days);
Fast pages, required checks on the branch, self-hosted runners, honest incidents2030 // On a self-hosted runner, `runner` and `RUNNER_*` describe that
2031 // machine rather than g1t's sandbox.
2032 let mut variables = info.variables(&job.key);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22033 variables.insert("GITHUB_RETENTION_DAYS".into(), json!(retention_days.to_string()));
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2034 let mut runner = match &job.runner_id {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2035 Some(id) => self.runner_context_for(id, &mut variables).await?,
2036 None => runner_context(),
2037 };
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2038 // A re-run with debug logging: what GitHub sets for one.
2039 if run.debug != 0 {
2040 debug_logging(&mut variables, &mut runner);
2041 }
GitHub Actions on g1t, part two: running workflows2042
2043 // Where to check out: a pull request's fork, or the repository.
2044 let clone_url = match run.pull {
2045 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" => {
2046 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
2047 &self.work,
2048 "get_pull",
2049 &g1t_contracts::work::ViewArgs {
2050 repo: repo.clone(),
2051 number,
2052 viewer: self.workspace_actor(&repo.namespace).await?,
2053 after_seq: 0,
2054 },
2055 )
2056 .await?;
2057 match located {
2058 Outcome::Ok(detail) => match detail.pull.fork {
2059 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
2060 None => format!("{SITE}/{}.git", run.repo),
2061 },
2062 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
2063 }
2064 }
2065 _ => format!("{SITE}/{}.git", run.repo),
2066 };
2067
2068 Ok(Outcome::Ok(json!({
2069 "job": job.id,
2070 "run": run.id,
2071 "key": job.key,
2072 "name": job.name,
2073 "spec": spec.raw,
2074 "workflow": {
2075 "env": workflow.env,
2076 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
2077 },
2078 "github": github,
Fast pages, required checks on the branch, self-hosted runners, honest incidents2079 "variables": variables,
GitHub Actions on g1t, part two: running workflows2080 "event": info.event,
2081 "contexts": {
2082 "vars": vars,
2083 "secrets": secrets,
Actions: reusable workflows in the repository2084 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
GitHub Actions on g1t, part two: running workflows2085 "matrix": matrix,
2086 "needs": needs,
2087 "strategy": {
2088 "fail-fast": spec.fail_fast,
2089 "job-index": job.ordinal,
2090 "job-total": siblings,
2091 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
2092 },
Fast pages, required checks on the branch, self-hosted runners, honest incidents2093 "runner": runner,
GitHub Actions on g1t, part two: running workflows2094 },
2095 "checkout": {
2096 "repository": run.repo,
2097 "url": clone_url,
2098 "sha": run.sha,
2099 "ref": run.git_ref,
2100 "token": token,
2101 },
2102 "timeoutMinutes": job.timeout_minutes,
2103 "masks": masks,
Merge branch 'worktree-agent-a3abfcce648e87dca'2104 // As the job's log lists them at its start.
2105 "permissions": permissions.listed().into_iter().map(|(name, access)| (name.to_owned(), json!(access.as_str()))).collect::<Map<String, Value>>(),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22106 // Whether the job may ask for an OIDC token decides whether it
2107 // is told where to.
2108 "runtime": {
2109 "token": runtime_token,
2110 "idToken": self.oidc_allowed(&run, &job),
2111 },
GitHub Actions on g1t, part two: running workflows2112 })))
2113 }
2114
2115 /// `job_report`: the sandbox telling how the job is going.
2116 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2117 let job = check!(self.job_for_token(&a).await?);
2118 let report = &a.report;
2119 let at = now();
2120 match report["kind"].as_str().unwrap_or_default() {
2121 "steps" => {
2122 // The list can grow as the job goes (post steps), so steps
2123 // already reported keep where they stand.
2124 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2125 let steps: Vec<Value> = report["steps"]
2126 .as_array()
2127 .map(|names| {
2128 names
2129 .iter()
2130 .enumerate()
2131 .map(|(i, name)| match known.get(i) {
2132 Some(step) if step["status"] != "queued" => step.clone(),
2133 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
2134 })
2135 .collect()
2136 })
2137 .unwrap_or_default();
2138 self.db
2139 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2140 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2141 .run()
2142 .await?;
2143 }
2144 "step" => {
2145 let number = report["number"].as_u64().unwrap_or(0) as usize;
2146 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2147 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
2148 let status = report["status"].as_str().unwrap_or("in_progress");
2149 step["status"] = json!(status);
2150 if status == "in_progress" {
2151 step["startedAt"] = json!(at);
2152 }
2153 if status == "completed" {
2154 step["finishedAt"] = json!(at);
2155 step["conclusion"] = report["conclusion"].clone();
2156 }
2157 if let Some(name) = report["name"].as_str() {
2158 step["name"] = json!(name);
2159 }
2160 }
2161 self.db
2162 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2163 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2164 .run()
2165 .await?;
2166 }
2167 "log" => {
2168 let mut text = report["text"].as_str().unwrap_or_default().to_owned();
2169 if text.len() > MAX_CHUNK_BYTES {
2170 let mut cut = MAX_CHUNK_BYTES;
2171 while !text.is_char_boundary(cut) {
2172 cut -= 1;
2173 }
2174 text.truncate(cut);
2175 }
2176 #[derive(Deserialize)]
2177 struct Size {
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs2178 n: Option<f64>,
2179 seq: Option<f64>,
GitHub Actions on g1t, part two: running workflows2180 }
2181 let size = self
2182 .db
2183 .prepare("SELECT SUM(LENGTH(text)) AS n, MAX(seq) AS seq FROM logs WHERE job_id = ?")
2184 .bind(&[job.id.as_str().into()])?
2185 .first::<Size>(None)
2186 .await?;
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs2187 let (used, seq) = size.map_or((0, 0.0), |s| (s.n.unwrap_or(0.0) as usize, s.seq.unwrap_or(0.0)));
GitHub Actions on g1t, part two: running workflows2188 if used < MAX_LOG_BYTES {
2189 if used + text.len() >= MAX_LOG_BYTES {
2190 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
2191 }
2192 self.db
2193 .prepare("INSERT INTO logs (job_id, seq, step, text) VALUES (?, ?, ?, ?)")
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs2194 .bind(&[job.id.as_str().into(), // Numbers go to D1 as f64: a u64 would be a BigInt, which it refuses.
2195 (seq + 1.0).into(), (report["step"].as_u64().unwrap_or(0) as u32).into(), text.into()])?
GitHub Actions on g1t, part two: running workflows2196 .run()
2197 .await?;
2198 }
2199 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2200 }
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2201 "summary" => {
2202 // `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
2203 // step's summary, up to 1 MiB a step and 20 steps a job.
2204 let step = report["step"].as_u64().unwrap_or(0) as u32;
2205 let markdown = report["markdown"].as_str().unwrap_or_default();
2206 #[derive(Deserialize)]
2207 struct Held {
2208 steps: u32,
2209 mine: Option<f64>,
2210 }
2211 let held = self
2212 .db
2213 .prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
2214 .bind(&[step.into(), job.id.as_str().into()])?
2215 .first::<Held>(None)
2216 .await?;
2217 let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));
2218 if summary_fits(steps, mine, markdown.len()) {
2219 self.db
2220 .prepare(
2221 "INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
2222 ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",
2223 )
2224 .bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
2225 .run()
2226 .await?;
2227 }
2228 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2229 }
GitHub Actions on g1t, part two: running workflows2230 "annotation" => {
2231 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2232 if annotations.len() < MAX_ANNOTATIONS {
2233 annotations.push(json!({
2234 "level": report["level"].as_str().unwrap_or("notice"),
2235 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
2236 "title": report["title"],
2237 "file": report["file"],
2238 "line": report["line"],
2239 }));
2240 self.db
2241 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
2242 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
2243 .run()
2244 .await?;
2245 }
2246 }
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2247 // Nothing to tell; the answer says whether to stop.
2248 "ping" => {
2249 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2250 }
GitHub Actions on g1t, part two: running workflows2251 "done" => {
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2252 // A job told to stop ends cancelled, however its cleanup went.
2253 let conclusion = if job.cancel_requested_at.is_some() {
2254 "cancelled"
2255 } else {
2256 report["conclusion"]
2257 .as_str()
2258 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
2259 .unwrap_or("failure")
2260 };
GitHub Actions on g1t, part two: running workflows2261 let outputs = report["outputs"].as_object().cloned();
2262 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2263 }
2264 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2265 }
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2266 // `cancelled`: the run was cancelled, and the runner should stop
2267 // the step it is on and run only its cleanup steps.
2268 Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
GitHub Actions on g1t, part two: running workflows2269 }
2270
2271 // --- Every minute ---------------------------------------------------------------
2272
2273 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
2274 let minute = now_ms / 60_000 * 60_000;
2275 if let Err(error) = self.run_schedules(minute).await {
2276 worker::console_error!("actions: schedules failed: {error}");
2277 }
2278 // Jobs whose sandbox went quiet or ran past their time.
2279 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2280 for job in running {
2281 // Times in g1t's format compare as text.
2282 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
2283 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
2284 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2285 let over = job.started_at.as_deref().is_some_and(|started| started < before(limit).as_str());
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2286 let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());
2287 if cancel_overdue {
2288 // Cancelled, and still going after its grace period.
2289 self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
2290 self.advance(&job.run_id).await?;
2291 } else if over {
GitHub Actions on g1t, part two: running workflows2292 let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);
Fast pages, required checks on the branch, self-hosted runners, honest incidents2293 // A self-hosted runner is told to stop on its next poll.
2294 if job.runner_id.is_none() {
2295 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2296 }
GitHub Actions on g1t, part two: running workflows2297 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2298 } else if silent {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2299 let reason = match &job.runner_name {
2300 Some(name) => format!("The self-hosted runner {name} stopped answering."),
2301 None => "The runner stopped answering.".to_owned(),
2302 };
2303 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
GitHub Actions on g1t, part two: running workflows2304 }
2305 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2306 if let Err(error) = self.sweep_runners(now_ms).await {
2307 worker::console_error!("actions: the runners' sweep failed: {error}");
2308 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2309 // Jobs held at an environment whose wait timer has run out.
2310 if let Err(error) = self.release_gates().await {
2311 worker::console_error!("actions: environments' gates failed: {error}");
2312 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2313 // Once an hour: the cache's expired entries, and its storage.
2314 if (now_ms / 60_000) % 60 == 7
2315 && let Err(error) = self.sweep_cache(now_ms).await
2316 {
2317 worker::console_error!("actions: the cache's sweep failed: {error}");
2318 }
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22319 // And artifacts past their time, and the toolkit's abandoned parts.
2320 if (now_ms / 60_000) % 60 == 37 {
2321 if let Err(error) = self.sweep_artifacts(now_ms).await {
2322 worker::console_error!("actions: the artifacts' sweep failed: {error}");
2323 }
2324 if let Err(error) = self.sweep_blob_parts(now_ms).await {
2325 worker::console_error!("actions: the blob parts' sweep failed: {error}");
2326 }
2327 }
GitHub Actions on g1t, part two: running workflows2328 self.start_queued().await
2329 }
2330}
2331
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look2332
2333#[cfg(test)]
2334mod stopping {
2335 use super::stops_runs;
2336 use g1t_contracts::events::Event;
2337 use serde_json::{Value, json};
2338
2339 fn event(kind: &str, data: Value) -> Event {
2340 Event {
2341 id: "evt_1".into(),
2342 kind: kind.into(),
2343 source: "repos".into(),
2344 time: "2026-10-05T00:00:00Z".into(),
2345 repo_id: Some("rep_1".into()),
2346 actor: None,
2347 data,
2348 }
2349 }
2350
2351 #[test]
2352 fn deleting_or_archiving_stops_runs() {
2353 assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));
2354 assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));
2355 assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);
2356 assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);
2357 assert_eq!(stops_runs(&event("git.push", json!({}))), None);
2358 }
2359}
Fast pages, required checks on the branch, self-hosted runners, honest incidents2360
2361#[cfg(test)]
2362mod status_of_needs {
2363 use std::collections::HashMap;
2364
2365 use super::ancestor_failed;
2366
2367 /// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,
2368 /// and a job that needs only the last.
2369 fn graph() -> HashMap<&'static str, Vec<&'static str>> {
2370 HashMap::from([
2371 ("check", vec![]),
2372 ("plan", vec!["check"]),
2373 ("migrate", vec!["plan"]),
2374 ("core", vec!["plan", "migrate"]),
2375 ("edge", vec!["plan", "migrate", "core"]),
2376 ("notify", vec!["edge"]),
2377 ])
2378 }
2379
2380 #[test]
2381 fn a_failure_is_seen_however_far_back() {
2382 let needs = graph();
2383 let failed = |which: &'static str| move |key: &str| key == which;
2384 // check failed; plan, and everything after, was skipped for it.
2385 assert!(ancestor_failed(&needs, "notify", failed("check")));
2386 assert!(ancestor_failed(&needs, "core", failed("check")));
2387 assert!(ancestor_failed(&needs, "edge", failed("core")));
2388 // Nothing before a job failed: a skipped migrate is not a failure.
2389 assert!(!ancestor_failed(&needs, "edge", |_| false));
2390 assert!(!ancestor_failed(&needs, "core", failed("edge")));
2391 assert!(!ancestor_failed(&needs, "check", failed("check")));
2392 }
2393
2394 #[test]
2395 fn cycles_and_unknown_keys_end() {
2396 let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);
2397 assert!(!ancestor_failed(&needs, "a", |_| false));
2398 assert!(!ancestor_failed(&needs, "missing", |_| true));
2399 }
2400}
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb972401
2402#[cfg(test)]
2403mod deployments {
2404 use serde_json::{Map, Value, json};
2405
2406 use super::{JobEnvironment, deployment_outcome, environment_of};
2407
2408 #[test]
2409 fn a_jobs_environment_is_read_for_deployments() {
2410 let contexts: Map<String, Value> = serde_json::from_value(json!({
2411 "github": { "ref_name": "main", "repository": "acme/web" },
2412 "inputs": { "target": "staging" },
2413 "matrix": {},
2414 }))
2415 .unwrap();
2416 let read = |raw: Value| environment_of(&raw, &contexts);
2417 assert_eq!(read(json!({})), None);
2418 assert_eq!(
2419 read(json!({ "environment": "production" })),
2420 Some(JobEnvironment { name: "production".into(), url: None, deploys: true })
2421 );
2422 assert_eq!(
2423 read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),
2424 Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })
2425 );
2426 // Expressions are filled in from the run.
2427 assert_eq!(
2428 read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),
2429 Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })
2430 );
2431 // Secrets only: no deployment.
2432 assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);
2433 // Only http(s) addresses.
2434 assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);
2435 }
2436
2437 #[test]
2438 fn a_runs_outcome_for_an_environment() {
2439 let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());
2440 assert_eq!(of(&["success", "skipped"]), Some("success"));
2441 assert_eq!(of(&["success", "failure"]), Some("failure"));
2442 assert_eq!(of(&["success", "cancelled"]), Some("error"));
2443 assert_eq!(of(&["skipped"]), None);
2444 assert_eq!(deployment_outcome(&[None]), None);
2445 }
2446}
Actions: keep every run attempt, re-run one job, graceful cancel, summaries2447
2448#[cfg(test)]
2449mod reruns {
2450 use serde_json::{Map, Value, json};
2451
2452 use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, debug_logging, rerun_keys, summary_fits, top_key};
2453
2454 /// build ← test ← deploy, and lint on its own.
2455 fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
2456 let order = ["build", "lint", "test", "deploy"];
2457 let needs = |key: &str| -> Vec<String> {
2458 match key {
2459 "test" => vec!["build".into()],
2460 "deploy" => vec!["test".into()],
2461 _ => Vec::new(),
2462 }
2463 };
2464 rerun_keys(&order, needs, |key| !failed.contains(&key), which)
2465 }
2466
2467 #[test]
2468 fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
2469 assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
2470 assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
2471 assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
2472 assert!(keys(Rerun::Failed, &[]).is_empty());
2473 // One job, whatever it came to, and what depends on it.
2474 assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
2475 assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
2476 assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
2477 assert!(keys(Rerun::Job("missing"), &[]).is_empty());
2478 }
2479
2480 #[test]
2481 fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
2482 assert_eq!(top_key("build/test"), "build");
2483 assert_eq!(top_key("build/inner/test"), "build");
2484 assert_eq!(top_key("lint"), "lint");
2485 }
2486
2487 #[test]
2488 fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
2489 assert!(summary_fits(0, None, 10));
2490 assert!(!summary_fits(0, None, 0), "nothing to keep");
2491 assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
2492 assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
2493 assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
2494 assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
2495 assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
2496 }
2497
2498 #[test]
2499 fn a_debug_rerun_sets_what_github_sets() {
2500 let mut variables = Map::new();
2501 let mut runner = json!({ "name": "g1t", "debug": "" });
2502 debug_logging(&mut variables, &mut runner);
2503 assert_eq!(variables["RUNNER_DEBUG"], "1");
2504 assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
2505 assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
2506 assert_eq!(runner["debug"], Value::String("1".into()));
2507 assert_eq!(runner["name"], "g1t");
2508 }
2509}

This file's history is long; its oldest lines are credited to the oldest commit read.