Skip to content
Sign up

flagon-io/g1t

public
0
2,533 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;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)32/// 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,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)118 /// Who started the current attempt, once it is re-run (migration 0009).
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,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)218 /// When it was told to stop, while it runs its cleanup steps
219 /// (migration 0009).
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.
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)651 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
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1089 /// A job that calls a reusable workflow, in the repository or another
1090 /// (reach.rs): that workflow's jobs join the run under it, with the
1091 /// inputs and secrets it passes.
Actions: reusable workflows in the repository1092 async fn call_workflow(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, uses: &str, scope: &Scope<'_>) -> Result<()> {
1093 let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;
1094 if depth > MAX_CALL_DEPTH {
1095 return self.fail_job(row, &format!("Reusable workflows call each other more than {MAX_CALL_DEPTH} deep.")).await;
1096 }
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1097 let (file, source, origin) = match self.called_workflow(run, row, uses).await? {
1098 Ok(found) => found,
1099 Err(why) => return self.fail_job(row, &why).await,
Actions: reusable workflows in the repository1100 };
1101 let called = match workflow::parse(&source) {
1102 Ok(called) => called,
1103 Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,
1104 };
1105 let Some(trigger) = called.trigger("workflow_call") else {
1106 return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;
1107 };
1108 // Inputs: what the caller passes, else the called workflow's defaults.
1109 let given = match job.raw.get("with") {
1110 Some(with) => match expr::interpolate_value(with, scope) {
1111 Ok(Value::Object(given)) => given,
1112 Ok(_) => Map::new(),
1113 Err(problem) => return self.fail_job(row, &format!("Its `with` does not read: {problem}")).await,
1114 },
1115 None => Map::new(),
1116 };
1117 let mut inputs = Map::new();
1118 for (name, spec) in &trigger.inputs {
1119 let value = given.get(name).cloned().or_else(|| spec.get("default").cloned()).unwrap_or(Value::Null);
1120 if value.is_null() && spec.get("required").and_then(Value::as_bool) == Some(true) {
1121 return self.fail_job(row, &format!("`{file}` needs the input `{name}`.")).await;
1122 }
1123 inputs.insert(name.clone(), value);
1124 }
1125 for (name, value) in given {
1126 inputs.entry(name).or_insert(value);
1127 }
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1128 // Secrets: none but the job's token unless `secrets:` passes them,
1129 // by name or with `inherit`; read when each job starts (job_spec).
1130 let outer = row.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());
1131 let secrets = crate::reach::secrets_plan(&job.raw, scope.contexts, outer);
1132 if let Some(name) = crate::reach::missing_secrets(&called.raw, &secrets).first() {
1133 return self.fail_job(row, &format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`.")).await;
1134 }
Actions: reusable workflows in the repository1135 let mut statements = Vec::new();
1136 for called_job in &called.jobs {
1137 let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{}/{n}", row.key)).collect();
1138 let call = json!({
1139 "role": "callee", "parent": row.key, "job": called_job.id, "path": file,
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1140 "source": source, "inputs": inputs, "depth": depth, "origin": origin, "secrets": secrets,
Actions: reusable workflows in the repository1141 });
1142 statements.push(
1143 self.db
1144 .prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")
1145 .bind(&[
1146 new_id("job", now_ms()).into(),
1147 row.run_id.as_str().into(),
1148 row.repo_id.as_str().into(),
1149 row.namespace.as_str().into(),
1150 format!("{}/{}", row.key, called_job.id).into(),
1151 format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),
1152 serde_json::to_string(&needs)?.into(),
1153 serde_json::to_string(&call)?.into(),
1154 ])?,
1155 );
1156 }
1157 statements.push(
1158 self.db
1159 .prepare("UPDATE jobs SET status = 'calling', call = ?, reason = ?, started_at = ? WHERE id = ?")
1160 .bind(&[
1161 serde_json::to_string(&json!({ "role": "caller", "path": file, "source": source }))?.into(),
1162 format!("Calls `{file}`.").into(),
1163 now().into(),
1164 row.id.as_str().into(),
1165 ])?,
1166 );
1167 self.db.batch(statements).await?;
1168 Ok(())
1169 }
1170
1171 /// A job that called a workflow, finished with its jobs: their result,
1172 /// and the outputs the workflow declares.
1173 async fn finish_call(&self, row: &JobRow, children: &[&JobRow]) -> Result<()> {
1174 let call = row.call().unwrap_or_default();
1175 let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());
1176 let mut jobs_context = Map::new();
1177 let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();
1178 for child in children {
1179 by_key.entry(child.key.clone()).or_default().push(child);
1180 }
1181 for (key, rows) in &by_key {
1182 let mut outputs = Map::new();
1183 for child in rows {
1184 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {
1185 outputs.extend(more);
1186 }
1187 }
1188 let id = key.rsplit('/').next().unwrap_or(key);
1189 jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));
1190 }
1191 let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));
1192 let mut contexts = Map::new();
1193 contexts.insert("jobs".into(), Value::Object(jobs_context));
1194 contexts.insert("inputs".into(), inputs);
1195 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
1196 let mut outputs = Map::new();
1197 if let Some(Value::Object(declared)) = called.as_ref().map(|w| {
1198 let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);
1199 on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)
1200 }) {
1201 for (name, spec) in declared {
1202 if let Some(value) = spec.get("value") {
1203 let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);
1204 outputs.insert(name, Value::String(expr::to_text(&value)));
1205 }
1206 }
1207 }
1208 let conclusion = key_result(children);
1209 self.db
1210 .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling'")
1211 .bind(&[conclusion.into(), serde_json::to_string(&outputs)?.into(), now().into(), row.id.as_str().into()])?
1212 .run()
1213 .await?;
1214 Ok(())
1215 }
1216
1217 /// A file's text at a commit, if it is there.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1218 pub(crate) async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {
Actions: reusable workflows in the repository1219 let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(
1220 &self.repos,
1221 "blob",
1222 &g1t_contracts::repos::BlobArgs {
1223 path: path.clone(),
1224 viewer: Some(ws.clone()),
1225 git_ref: sha.to_owned(),
1226 file_path: file.to_owned(),
1227 },
1228 )
1229 .await?;
1230 Ok(match blob {
1231 Outcome::Ok(view) => view.text,
1232 Outcome::Fail(_) => None,
1233 })
1234 }
1235
GitHub Actions on g1t, part two: running workflows1236 async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {
1237 self.db
1238 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")
1239 .bind(&[optional(reason), now().into(), row.id.as_str().into()])?
1240 .run()
1241 .await?;
1242 Ok(())
1243 }
1244
1245 async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {
1246 self.db
1247 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")
1248 .bind(&[reason.into(), now().into(), row.id.as_str().into()])?
1249 .run()
1250 .await?;
1251 Ok(())
1252 }
1253
1254 /// Starts queued jobs, oldest first, while their workspace has room.
1255 pub async fn start_queued(&self) -> Result<()> {
1256 let queued = self
1257 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1258 // Self-hosted jobs are taken by their runners (runners.rs).
1259 .prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NULL ORDER BY rowid LIMIT 50")
GitHub Actions on g1t, part two: running workflows1260 .all()
1261 .await?
1262 .results::<JobRow>()?;
1263 let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
1264 for job in queued {
1265 let in_workspace = match running.get(&job.namespace) {
1266 Some(n) => *n,
1267 None => {
1268 let n = self
1269 .db
Fast pages, required checks on the branch, self-hosted runners, honest incidents1270 // Only g1t's own sandboxes count against the workspace's room.
1271 .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 workflows1272 .bind(&[job.namespace.as_str().into()])?
1273 .first::<Count>(None)
1274 .await?
1275 .map_or(0, |count| count.n);
1276 running.insert(job.namespace.clone(), n);
1277 n
1278 }
1279 };
1280 if in_workspace >= RUNNING_PER_WORKSPACE {
1281 continue;
1282 }
1283 if let Some(max) = job.max_parallel {
1284 let siblings = self
1285 .db
1286 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
1287 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1288 .first::<Count>(None)
1289 .await?
1290 .map_or(0, |count| count.n);
1291 if siblings >= max {
1292 continue;
1293 }
1294 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1295 // One job of a concurrency group runs at a time.
1296 if let Some(group) = &job.concurrency_group {
1297 let running = self
1298 .db
1299 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'in_progress' AND id != ?")
1300 .bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])?
1301 .first::<Count>(None)
1302 .await?
1303 .map_or(0, |count| count.n);
1304 if running > 0 {
1305 continue;
1306 }
1307 }
GitHub Actions on g1t, part two: running workflows1308 let token = random_hex(24);
1309 let at = now();
1310 let claimed = self
1311 .db
1312 .prepare(
1313 "UPDATE jobs SET status = 'in_progress', token_hash = ?, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
1314 )
1315 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1316 .first::<Value>(None)
1317 .await?;
1318 if claimed.is_none() {
1319 continue;
1320 }
1321 running.insert(job.namespace.clone(), in_workspace + 1);
1322 self.db
1323 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
1324 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?
1325 .run()
1326 .await?;
1327 let run = self.run_row(&job.run_id).await?;
1328 let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {
1329 namespace: job.namespace.clone(),
1330 name: String::new(),
1331 });
Fast pages, required checks on the branch, self-hosted runners, honest incidents1332 // Its environment and the machine it asked for, for the runner.
1333 let details = run.as_ref().map(|run| start_details(run, &job)).unwrap_or_default();
GitHub Actions on g1t, part two: running workflows1334 let started: Outcome<Value> = g1t_kit::call(
1335 &self.runner,
1336 "start_actions_job",
1337 &StartJobArgs {
1338 job: job.id.clone(),
1339 token,
1340 repo,
1341 timeout_minutes: job.timeout_minutes,
Fast pages, required checks on the branch, self-hosted runners, honest incidents1342 workflow: run.as_ref().map(|run| run.path.clone()),
1343 environment: details.environment,
1344 trusted: run.as_ref().is_some_and(|run| run.trusted != 0),
1345 instance: details.instance,
GitHub Actions on g1t, part two: running workflows1346 },
1347 )
1348 .await
1349 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
1350 if let Outcome::Fail(refused) = started {
Merge branch 'worktree-agent-ab9543c492a7ed481' into spend-guardrails1351 // Billing would not pay for a scheduled run's job: the
1352 // schedule waits rather than making runs only to refuse them.
1353 if refused.code == FailureCode::PaymentRequired
1354 && let Some(run) = run.as_ref().filter(|run| run.event == "schedule")
1355 {
1356 self.db
1357 .prepare("UPDATE workflows SET schedule_refused_until = ? WHERE id = ?")
1358 .bind(&[rfc3339(now_ms() + crate::SCHEDULE_REFUSED_MS).into(), run.workflow_id.as_str().into()])?
1359 .run()
1360 .await?;
1361 }
GitHub Actions on g1t, part two: running workflows1362 Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971363 } else {
1364 self.job_started(&job.id).await?;
GitHub Actions on g1t, part two: running workflows1365 }
1366 }
1367 Ok(())
1368 }
1369
1370 /// Finishes a job and moves its run along.
1371 pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {
1372 let finished = self
1373 .db
1374 .prepare(
1375 "UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
1376 finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
1377 )
1378 .bind(&[
1379 conclusion.into(),
1380 optional(reason),
1381 outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
1382 now().into(),
1383 job_id.into(),
1384 ])?
1385 .first::<JobRow>(None)
1386 .await?;
1387 let Some(job) = finished else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1388 // Its G1T_TOKEN stops working with it.
1389 self.revoke_job_tokens(&job.id).await;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971390 // A job that deploys failed: so did its run's deployment, now.
1391 if conclusion == "failure" && job.continue_on_error == 0 && job.started_at.is_some()
1392 && let Some(run) = self.run_row(&job.run_id).await?
1393 && let Some(env) = deploys_to(&run, &job)
1394 {
1395 self.report_deployment(&run, &env, "failure", false).await;
1396 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1397 // A self-hosted runner's job: the runner is free again, and its time
1398 // is recorded, at nothing.
1399 if job.runner_id.is_some() {
1400 self.released(&job).await?;
1401 }
GitHub Actions on g1t, part two: running workflows1402 // Steps still marked as going are not going any more.
1403 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1404 let mut touched = false;
1405 for step in steps.iter_mut() {
1406 if step["status"] != "completed" {
1407 let was_running = step["status"] == "in_progress";
1408 step["status"] = json!("completed");
1409 step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });
1410 touched = true;
1411 }
1412 }
1413 if touched {
1414 self.db
1415 .prepare("UPDATE jobs SET steps = ? WHERE id = ?")
1416 .bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?
1417 .run()
1418 .await?;
1419 }
1420 // fail-fast: one failed combination stops the rest of its matrix.
1421 if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {
1422 let run = self.run_row(&job.run_id).await?;
1423 let fail_fast = run
1424 .as_ref()
1425 .and_then(|run| workflow::parse(&run.source).ok())
1426 .and_then(|workflow| workflow.jobs.into_iter().find(|j| j.id == job.key))
1427 .is_none_or(|j| j.fail_fast);
1428 if fail_fast {
1429 let siblings = self
1430 .db
1431 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
1432 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
1433 .all()
1434 .await?
1435 .results::<JobRow>()?;
1436 for sibling in siblings {
1437 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
1438 }
1439 }
1440 }
1441 Box::pin(self.advance(&job.run_id)).await
1442 }
1443
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1444 /// Cancels a job. One that is running is told to stop (the answer to
1445 /// its next report), and runs its `if: always()` and `cancelled()`
1446 /// steps and its post steps before it ends `cancelled`; one that has
1447 /// not started is cancelled at once.
GitHub Actions on g1t, part two: running workflows1448 async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1449 if job.status == "in_progress" && job.started_at.is_some() {
1450 self.db
1451 .prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")
1452 .bind(&[now().into(), reason.into(), job.id.as_str().into()])?
1453 .run()
1454 .await?;
1455 return Ok(());
1456 }
1457 self.hard_stop(job, reason).await
1458 }
1459
1460 /// Cancels a job outright, stopping its sandbox if it has one.
1461 async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {
Fast pages, required checks on the branch, self-hosted runners, honest incidents1462 // A self-hosted runner hears it was cancelled on its next poll.
1463 if job.status == "in_progress" && job.runner_id.is_none() {
GitHub Actions on g1t, part two: running workflows1464 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
1465 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1466 let stopped = self
1467 .db
1468 .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 workflows1469 .bind(&[reason.into(), now().into(), job.id.as_str().into()])?
Fast pages, required checks on the branch, self-hosted runners, honest incidents1470 .first::<JobRow>(None)
GitHub Actions on g1t, part two: running workflows1471 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'1472 if stopped.as_ref().is_some_and(|row| row.started_at.is_some()) {
1473 self.revoke_job_tokens(&job.id).await;
1474 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents1475 if let Some(stopped) = stopped.filter(|row| row.runner_id.is_some()) {
1476 self.released(&stopped).await?;
1477 }
GitHub Actions on g1t, part two: running workflows1478 Ok(())
1479 }
1480
Merge branch 'worktree-agent-a3abfcce648e87dca'1481 /// Ends a job's tokens at once. A failure is logged: the token expires
1482 /// on its own soon after the job's time limit.
1483 async fn revoke_job_tokens(&self, job_id: &str) {
1484 let revoked: Result<bool> =
1485 g1t_kit::call(&self.identity, "revoke_job_tokens", &RevokeJobTokensArgs { job_id: job_id.to_owned() }).await;
1486 if let Err(error) = revoked {
1487 worker::console_error!("actions: the tokens of job {job_id} were not revoked: {error}");
1488 }
1489 }
1490
GitHub Actions on g1t, part two: running workflows1491 /// Finishes the run when every job has.
1492 async fn finish_if_done(&self, run_id: &str) -> Result<()> {
1493 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
Merge branch 'worktree-agent-a3abfcce648e87dca'1494 if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {
GitHub Actions on g1t, part two: running workflows1495 return Ok(());
1496 }
1497 let jobs = self.job_rows(run_id).await?;
1498 if !jobs.iter().all(|job| job.status == "completed") {
1499 return Ok(());
1500 }
1501 self.finish_run(&run, None).await
1502 }
1503
1504 async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {
1505 let jobs = self.job_rows(&run.id).await?;
1506 let rows: Vec<&JobRow> = jobs.iter().collect();
1507 let conclusion = if error.is_some() {
1508 "failure"
1509 } else if run.conclusion.as_deref() == Some("cancelled") {
1510 "cancelled"
1511 } else if rows.is_empty() {
1512 "skipped"
1513 } else {
1514 key_result(&rows)
1515 };
1516 let done = self
1517 .db
1518 .prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")
1519 .bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?
1520 .first::<Value>(None)
1521 .await?;
1522 if done.is_none() {
1523 return Ok(());
1524 }
1525 self.report_status(run, conclusion).await?;
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971526 self.settle_deployments(run, &jobs, conclusion == "cancelled").await;
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 241527 let published: Result<()> = g1t_kit::call(
1528 &self.events,
1529 "publish",
1530 &g1t_contracts::events::Publish {
1531 events: vec![g1t_contracts::events::NewEvent {
1532 kind: "workflow.completed",
1533 source: "actions",
1534 repo_id: Some(run.repo_id.clone()),
1535 actor: run.actor_id.clone(),
1536 data: g1t_contracts::events::WorkflowEvent {
1537 run_id: run.id.clone(),
1538 repo_id: run.repo_id.clone(),
1539 workflow: run.name.clone(),
1540 path: run.path.clone(),
1541 number: run.number,
1542 event: run.event.clone(),
1543 conclusion: conclusion.to_owned(),
1544 git_ref: run.git_ref.clone(),
1545 sha: run.sha.clone(),
1546 pull: run.pull,
1547 },
1548 }],
1549 },
1550 )
1551 .await;
1552 if let Err(error) = published {
1553 worker::console_error!("actions: could not publish workflow.completed: {error}");
1554 }
GitHub Actions on g1t, part two: running workflows1555 // The next run waiting in its concurrency group.
1556 if let Some(group) = &run.concurrency_group {
1557 let next = self
1558 .db
1559 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")
1560 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1561 .first::<RunRow>(None)
1562 .await?;
1563 if let Some(next) = next {
1564 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;
1565 Box::pin(self.advance(&next.id)).await?;
1566 }
1567 }
1568 Ok(())
1569 }
1570
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb971571 /// Tells the deployments service how a run's deployment to `env` stands:
1572 /// made when its first job naming the environment starts, failed when one
1573 /// of them fails, and settled (`last`) when the run finishes. Never fails
1574 /// the run: a deployment that could not be recorded is logged.
1575 async fn report_deployment(&self, run: &RunRow, env: &JobEnvironment, state: &str, last: bool) {
1576 let path = repo_path(&run.repo);
1577 let reported: Result<Value> = g1t_kit::call(
1578 &self.deployments,
1579 "actions_deployment",
1580 &json!({
1581 "repoId": run.repo_id,
1582 "repo": { "namespace": path.namespace, "name": path.name },
1583 "runId": run.id,
1584 "attempt": run.attempt,
1585 "runUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1586 "environment": env.name,
1587 "url": env.url,
1588 "ref": run.git_ref,
1589 "sha": run.sha,
1590 "state": state,
1591 "final": last,
1592 "creator": run.actor,
1593 "workflow": run.name,
1594 }),
1595 )
1596 .await;
1597 if let Err(error) = reported {
1598 worker::console_error!("actions: deployment to {} not recorded for run {}: {error}", env.name, run.id);
1599 }
1600 }
1601
1602 /// A job that deploys has started: its run's deployment to the
1603 /// environment is under way.
1604 pub(crate) async fn job_started(&self, job_id: &str) -> Result<()> {
1605 let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[job_id.into()])?.first::<JobRow>(None).await? else {
1606 return Ok(());
1607 };
1608 let Some(run) = self.run_row(&job.run_id).await? else { return Ok(()) };
1609 if let Some(env) = deploys_to(&run, &job) {
1610 self.report_deployment(&run, &env, "in_progress", false).await;
1611 }
1612 Ok(())
1613 }
1614
1615 /// The run is over: each environment its jobs deployed to takes the
1616 /// outcome of those jobs (`deployment_outcome`).
1617 async fn settle_deployments(&self, run: &RunRow, jobs: &[JobRow], cancelled: bool) {
1618 let mut seen: Vec<(JobEnvironment, Vec<Option<String>>)> = Vec::new();
1619 for job in jobs {
1620 let Some(env) = deploys_to(run, job) else { continue };
1621 let conclusion = if cancelled && job.conclusion.as_deref() != Some("skipped") && job.started_at.is_some() {
1622 Some("cancelled".to_owned())
1623 } else if job.started_at.is_none() {
1624 Some("skipped".to_owned())
1625 } else {
1626 job.conclusion.clone()
1627 };
1628 match seen.iter_mut().find(|(known, _)| known.name.eq_ignore_ascii_case(&env.name)) {
1629 Some((known, conclusions)) => {
1630 if known.url.is_none() {
1631 known.url = env.url.clone();
1632 }
1633 conclusions.push(conclusion);
1634 }
1635 None => seen.push((env, vec![conclusion])),
1636 }
1637 }
1638 for (env, conclusions) in seen {
1639 if let Some(state) = deployment_outcome(&conclusions) {
1640 self.report_deployment(run, &env, state, true).await;
1641 }
1642 }
1643 }
1644
GitHub Actions on g1t, part two: running workflows1645 /// Tells the pull request (or commit) how the run went, as a status.
1646 async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {
1647 let state = match conclusion {
1648 "success" | "skipped" => "success",
1649 "cancelled" => "error",
1650 _ => "failure",
1651 };
1652 let _: Result<Value> = g1t_kit::call(
1653 &self.work,
1654 "set_commit_status",
1655 &json!({
1656 "repoId": run.repo_id,
1657 "sha": run.sha,
1658 "context": format!("{} / {}", run.name, run.event),
1659 "state": state,
1660 "description": format!("{} {}", run.name, match conclusion {
1661 "success" => "passed",
1662 "skipped" => "was skipped",
1663 "cancelled" => "was cancelled",
1664 _ => "failed",
1665 }),
1666 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1667 "source": "actions",
GitHub Actions on g1t, part two: running workflows1668 }),
1669 )
1670 .await;
1671 Ok(())
1672 }
1673
1674 /// Tells the pull request a run has started on its head.
1675 pub async fn report_pending(&self, run: &RunRow) -> Result<()> {
1676 let _: Result<Value> = g1t_kit::call(
1677 &self.work,
1678 "set_commit_status",
1679 &json!({
1680 "repoId": run.repo_id,
1681 "sha": run.sha,
1682 "context": format!("{} / {}", run.name, run.event),
1683 "state": "pending",
Merge branch 'worktree-agent-a3abfcce648e87dca'1684 "description": if run.status == "action_required" {
1685 format!("{} is waiting for approval", run.name)
1686 } else {
1687 format!("{} is running", run.name)
1688 },
GitHub Actions on g1t, part two: running workflows1689 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge1690 "source": "actions",
GitHub Actions on g1t, part two: running workflows1691 }),
1692 )
1693 .await;
1694 Ok(())
1695 }
1696
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1697 /// Cancels a run: its waiting and queued jobs at once, and its running
1698 /// ones gracefully (`stop_job`), or outright when `force`.
1699 pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {
GitHub Actions on g1t, part two: running workflows1700 self.db
1701 .prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
1702 .bind(&[run.id.as_str().into()])?
1703 .run()
1704 .await?;
1705 for job in self.job_rows(&run.id).await?.iter().filter(|job| job.status != "completed") {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1706 if force {
1707 self.hard_stop(job, reason).await?;
1708 } else {
1709 self.stop_job(job, reason).await?;
1710 }
GitHub Actions on g1t, part two: running workflows1711 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1712 if run.status == "pending" || run.status == "action_required" {
GitHub Actions on g1t, part two: running workflows1713 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
1714 }
1715 self.advance(&run.id).await
1716 }
1717
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1718 /// Cancels every run of the repository that has not finished, for
1719 /// `repo.deleted` and `repo.archived`.
1720 pub async fn stop_runs(&self, repo_id: &str) -> Result<()> {
1721 let runs = self
1722 .db
1723 .prepare("SELECT * FROM runs WHERE repo_id = ? AND status != 'completed'")
1724 .bind(&[repo_id.into()])?
1725 .all()
1726 .await?
1727 .results::<RunRow>()?;
1728 for run in runs {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1729 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 look1730 }
1731 Ok(())
1732 }
1733
GitHub Actions on g1t, part two: running workflows1734 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 look1735 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
GitHub Actions on g1t, part two: running workflows1736 return Ok(Outcome::Fail(refused));
1737 }
1738 let run = check!(self.run_in(&a.repo, &a.id).await?);
1739 if run.status == "completed" {
1740 return Ok(fail(FailureCode::Conflict, "The run has already finished."));
1741 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1742 // Cancelling a run that is already cancelling stops its jobs
1743 // outright, without waiting for their cleanup steps.
1744 let force = a.force || run.conclusion.as_deref() == Some("cancelled");
1745 let reason = if force {
1746 format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)
1747 } else {
1748 format!("{} cancelled the run.", a.actor.username)
1749 };
1750 self.cancel_run(&run, &reason, force).await?;
GitHub Actions on g1t, part two: running workflows1751 self.run_summary(&run.id).await
1752 }
1753
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1754 /// Runs again, as a new attempt: every job, with `failed_only` those
1755 /// that did not succeed, or with `job` that one; each with the jobs that
1756 /// need them. The attempt that ends is kept, its jobs and their logs and
1757 /// summaries with it (`job_attempts`, `run_attempts`).
GitHub Actions on g1t, part two: running workflows1758 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 look1759 if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {
GitHub Actions on g1t, part two: running workflows1760 return Ok(Outcome::Fail(refused));
1761 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1762 // A job alone (`POST …/jobs/{job}/rerun`) names its run.
1763 let run_id = match (&a.job, a.id.is_empty()) {
1764 (Some(job), true) => {
1765 #[derive(Deserialize)]
1766 struct Of {
1767 run_id: String,
1768 }
1769 let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;
1770 match of {
1771 Some(of) => of.run_id,
1772 None => return Ok(fail(FailureCode::NotFound, "No such job.")),
1773 }
1774 }
1775 _ => a.id.clone(),
1776 };
1777 let run = check!(self.run_in(&a.repo, &run_id).await?);
GitHub Actions on g1t, part two: running workflows1778 if run.status != "completed" {
1779 return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
1780 }
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1781 // Nothing starts again on an archived repository.
1782 match self.visible_repo(&a.repo, &Some(a.actor.clone())).await? {
1783 Some(repo) if repo.archived() => {
1784 return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));
1785 }
1786 Some(_) => {}
1787 None => return Ok(fail(FailureCode::NotFound, "There is no such repository.")),
1788 }
GitHub Actions on g1t, part two: running workflows1789 if run.error.is_some() {
1790 return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));
1791 }
1792 let jobs = self.job_rows(&run.id).await?;
1793 let workflow = workflow::parse(&run.source).ok();
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1794 let target = match &a.job {
1795 Some(id) => match jobs.iter().find(|job| &job.id == id) {
1796 Some(job) => Some(top_key(&job.key).to_owned()),
1797 None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),
1798 },
1799 None => None,
1800 };
1801 let which = match (&target, a.failed_only) {
1802 (Some(key), _) => Rerun::Job(key),
1803 (None, true) => Rerun::Failed,
1804 (None, false) => Rerun::All,
1805 };
1806 let order = workflow.as_ref().map(|w| w.job_order()).unwrap_or_default();
1807 let again = rerun_keys(
1808 &order,
1809 |key| jobs.iter().find(|j| j.key == key).map(JobRow::needs).unwrap_or_default(),
1810 |key| jobs.iter().filter(|j| j.key == key).all(|j| j.conclusion.as_deref() == Some("success")),
1811 which,
1812 );
GitHub Actions on g1t, part two: running workflows1813 if again.is_empty() {
1814 return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
1815 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1816 // The jobs that run again, a called workflow's with the job calling it.
1817 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();
1818 let ids = serde_json::to_string(&rerun_ids)?;
1819 let ended = run.attempt.to_string();
1820 let ended = ended.as_str();
1821 let mut statements = vec![
1822 // The attempt that ends, as it ended.
1823 self.db
1824 .prepare(
1825 "INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)
1826 SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",
1827 )
1828 .bind(&[run.id.as_str().into()])?,
1829 // Earlier attempts that showed a job's logs from where it ran
1830 // then now find them where they move to.
1831 self.db
1832 .prepare("UPDATE job_attempts SET log_id = log_id || '.' || ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")
1833 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
1834 // Each of its jobs: one that runs again keeps its logs under
1835 // `{id}.{attempt}`; one left alone is still the live job's.
1836 self.db
1837 .prepare(
1838 "INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,
1839 steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)
1840 SELECT id || '.' || ?1, run_id, repo_id, ?1, id,
1841 CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id || '.' || ?1 ELSE id END,
1842 key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at
1843 FROM jobs WHERE run_id = ?2 ORDER BY rowid",
1844 )
1845 .bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,
1846 self.db
1847 .prepare("UPDATE logs SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
1848 .bind(&[ended.into(), ids.as_str().into()])?,
1849 self.db
1850 .prepare("UPDATE job_summaries SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")
1851 .bind(&[ended.into(), ids.as_str().into()])?,
1852 ];
GitHub Actions on g1t, part two: running workflows1853 for key in &again {
1854 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 repository1855 // The jobs of a workflow it called are made again when it calls it again.
GitHub Actions on g1t, part two: running workflows1856 statements.push(
1857 self.db
Actions: reusable workflows in the repository1858 .prepare("DELETE FROM jobs WHERE run_id = ? AND key LIKE ?")
1859 .bind(&[run.id.as_str().into(), format!("{key}/%").into()])?,
1860 );
1861 statements.push(
1862 self.db
GitHub Actions on g1t, part two: running workflows1863 .prepare(
1864 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Fast pages, required checks on the branch, self-hosted runners, honest incidents1865 matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,
Merge branch 'worktree-agent-a3abfcce648e87dca'1866 labels = NULL, queued_at = NULL, runner_id = NULL, runner_name = NULL, environment = NULL,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1867 concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",
GitHub Actions on g1t, part two: running workflows1868 )
1869 .bind(&[run.id.as_str().into(), key.as_str().into()])?,
1870 );
1871 }
1872 statements.push(
1873 self.db
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)1874 .prepare(
1875 "UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,
1876 triggering_actor = ?, debug = ? WHERE id = ?",
1877 )
1878 .bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,
GitHub Actions on g1t, part two: running workflows1879 );
1880 self.db.batch(statements).await?;
1881 if let Some(run) = self.run_row(&run.id).await? {
1882 self.report_pending(&run).await?;
1883 }
1884 self.advance(&run.id).await?;
1885 self.run_summary(&run.id).await
1886 }
1887
1888 pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {
1889 let row = self
1890 .db
1891 .prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")
1892 .bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?
1893 .first::<RunRow>(None)
1894 .await?;
1895 Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))
1896 }
1897
1898 // --- The sandbox's side -----------------------------------------------------
1899
Fast pages, required checks on the branch, self-hosted runners, honest incidents1900 pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
GitHub Actions on g1t, part two: running workflows1901 let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
1902 Ok(match job {
1903 Some(job) if job.status == "in_progress" && job.token_hash.as_deref().is_some_and(|hash| same(hash, &sha256_hex(&a.token))) => {
1904 Outcome::Ok(job)
1905 }
1906 _ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
1907 })
1908 }
1909
A repository has its own sidebar, as settings do1910 /// `job_auth`: which run and repository a running job's token is for,
1911 /// so the API can keep its artifacts and cache.
1912 pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1913 let job = check!(self.job_for_token(&a).await?);
1914 Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))
1915 }
1916
GitHub Actions on g1t, part two: running workflows1917 /// `job_spec`: everything the sandbox needs to run the job.
1918 pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1919 let job = check!(self.job_for_token(&a).await?);
1920 let Some(run) = self.run_row(&job.run_id).await? else {
1921 return Ok(fail(FailureCode::NotFound, "No such run."));
1922 };
Actions: reusable workflows in the repository1923 let Ok(caller) = workflow::parse(&run.source) else {
GitHub Actions on g1t, part two: running workflows1924 return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));
1925 };
Actions: reusable workflows in the repository1926 // A called workflow's job runs as that workflow defines it.
1927 let callee = job.callee();
1928 let (workflow, spec, call_inputs) = match callee {
1929 Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),
1930 None => match caller.jobs.iter().find(|j| j.id == job.key) {
1931 Some(spec) => (caller.clone(), spec.clone(), None),
1932 None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),
1933 },
GitHub Actions on g1t, part two: running workflows1934 };
Actions: reusable workflows in the repository1935 let spec = &spec;
GitHub Actions on g1t, part two: running workflows1936 let repo = repo_path(&run.repo);
1937 let trusted = run.trusted != 0;
Merge branch 'worktree-agent-a3abfcce648e87dca'1938 // The job's `environment:`, by name, as read when its needs were done
1939 // (an expression included), once the environment's protection rules
1940 // let it start: entries with a value for it give that value instead
1941 // of their default, as GitHub's environment secrets do.
1942 let environment: Option<String> = job.environment.clone();
1943 // What its token may do: its `permissions:` (a called workflow's
1944 // jobs no more than the job that calls it), else the repository's
1945 // default; read-only for a pull request from outside.
1946 // The repository's default, its workspace's taken in, and whether
1947 // its jobs may open and approve pull requests.
1948 let (default, pull_requests) = self.token_policy(&run.repo_id).await?;
1949 let mut permissions = spec.permissions(&workflow, default);
1950 if let Some(parent) = &job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned)) {
1951 let top = parent.split('/').next().unwrap_or(parent);
1952 if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) {
1953 permissions = permissions.capped_by(&caller_job.permissions(&caller, default));
GitHub Actions on g1t, part two: running workflows1954 }
Merge branch 'worktree-agent-a3abfcce648e87dca'1955 }
1956 if !trusted {
1957 permissions = permissions.read_only();
1958 }
1959 // G1T_TOKEN, and GITHUB_TOKEN as its alias: a token of the
1960 // workspace's that reaches this repository only, with the scopes
1961 // its permissions give, until the job ends.
1962 let token = match self.workspace_actor(&repo.namespace).await? {
1963 Some(workspace) => {
1964 let created: CreatedAccessToken = g1t_kit::call(
1965 &self.identity,
1966 "create_job_token",
1967 &CreateJobTokenArgs {
1968 workspace,
1969 repo: repo.clone(),
1970 run_id: run.id.clone(),
1971 job_id: job.id.clone(),
1972 name: format!("G1T_TOKEN for {} run {}", run.repo, run.number),
1973 ttl_seconds: u64::from(job.timeout_minutes) * 60 + 600,
1974 scopes: permissions.scopes().into_iter().map(str::to_owned).collect(),
1975 pull_requests: pull_requests && trusted,
1976 },
1977 )
1978 .await?;
1979 created.token
1980 }
1981 None => String::new(),
GitHub Actions on g1t, part two: running workflows1982 };
Secrets and variables: one list, rows per environment, for workflows and deployments1983 // A run that is not trusted (a pull request from outside the
1984 // workspace) gets no secrets and an empty token.
Merge Actions: cross-repo workflows and actions, release and deployment triggers, step timeouts1985 let passed = job.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());
1986 let mut secrets = match (trusted, passed) {
1987 (false, _) => Map::new(),
1988 (true, None) => self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?,
1989 // A called workflow's job: what its callers passed it (reach.rs),
1990 // and its own environment's secrets over them.
1991 (true, Some(plan)) => {
1992 let base = self.secrets_for(&run.repo_id, &run.repo, None, true).await?;
1993 let vars = self.variables_for(&run.repo_id, &run.repo, None, true).await?;
1994 let github = run.info().context(&job.key, "", run.action.as_deref());
1995 let mut passed = crate::reach::resolve_secrets(&plan, &base, &github, &vars);
1996 if let Some(name) = environment.as_deref() {
1997 let own = self.secrets_for(&run.repo_id, &run.repo, Some(name), true).await?;
1998 for (key, value) in own {
1999 if base.get(&key) != Some(&value) {
2000 passed.insert(key, value);
2001 }
2002 }
2003 }
2004 passed
2005 }
Secrets and variables: one list, rows per environment, for workflows and deployments2006 };
2007 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
GitHub Actions on g1t, part two: running workflows2008 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
Merge branch 'worktree-agent-a3abfcce648e87dca'2009 // Each secret as it is, a line at a time, base64 and JSON-escaped.
2010 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 deployments2011 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 R22012 // The toolkit's runtime token (runtime.rs), for as long as the job
2013 // may run; the API puts it and the toolkit's addresses in the
2014 // job's variables.
2015 let runtime_token = crate::runtime::runtime_token(
2016 &job.id,
2017 &job.run_id,
2018 job.token_hash.as_deref().unwrap_or_default(),
2019 now_ms() / 1000,
2020 u64::from(job.timeout_minutes) * 60 + 600,
2021 );
2022 masks.push(runtime_token.clone());
2023 let retention_days = self.retention_setting(&run.repo_id).await?;
GitHub Actions on g1t, part two: running workflows2024
2025 let jobs = self.job_rows(&run.id).await?;
2026 let mut needs = Map::new();
Actions: reusable workflows in the repository2027 // In a called workflow, its jobs' keys sit under the job that called it.
2028 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 workflows2029 for need in &spec.needs {
Actions: reusable workflows in the repository2030 let key = match &parent {
2031 Some(parent) => format!("{parent}/{need}"),
2032 None => need.clone(),
2033 };
2034 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
GitHub Actions on g1t, part two: running workflows2035 let mut outputs = Map::new();
2036 for row in &rows {
2037 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
2038 outputs.extend(more);
2039 }
2040 }
2041 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
2042 }
2043 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
2044 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
2045 let info = run.info();
2046 let mut github = info.context(&job.key, &token, run.action.as_deref());
2047 github["token"] = json!(token);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22048 github["retention_days"] = json!(retention_days);
Fast pages, required checks on the branch, self-hosted runners, honest incidents2049 // On a self-hosted runner, `runner` and `RUNNER_*` describe that
2050 // machine rather than g1t's sandbox.
2051 let mut variables = info.variables(&job.key);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22052 variables.insert("GITHUB_RETENTION_DAYS".into(), json!(retention_days.to_string()));
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2053 let mut runner = match &job.runner_id {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2054 Some(id) => self.runner_context_for(id, &mut variables).await?,
2055 None => runner_context(),
2056 };
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2057 // A re-run with debug logging: what GitHub sets for one.
2058 if run.debug != 0 {
2059 debug_logging(&mut variables, &mut runner);
2060 }
GitHub Actions on g1t, part two: running workflows2061
2062 // Where to check out: a pull request's fork, or the repository.
2063 let clone_url = match run.pull {
2064 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" => {
2065 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
2066 &self.work,
2067 "get_pull",
2068 &g1t_contracts::work::ViewArgs {
2069 repo: repo.clone(),
2070 number,
2071 viewer: self.workspace_actor(&repo.namespace).await?,
2072 after_seq: 0,
2073 },
2074 )
2075 .await?;
2076 match located {
2077 Outcome::Ok(detail) => match detail.pull.fork {
2078 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
2079 None => format!("{SITE}/{}.git", run.repo),
2080 },
2081 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
2082 }
2083 }
2084 _ => format!("{SITE}/{}.git", run.repo),
2085 };
2086
2087 Ok(Outcome::Ok(json!({
2088 "job": job.id,
2089 "run": run.id,
2090 "key": job.key,
2091 "name": job.name,
2092 "spec": spec.raw,
2093 "workflow": {
2094 "env": workflow.env,
2095 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
2096 },
2097 "github": github,
Fast pages, required checks on the branch, self-hosted runners, honest incidents2098 "variables": variables,
GitHub Actions on g1t, part two: running workflows2099 "event": info.event,
2100 "contexts": {
2101 "vars": vars,
2102 "secrets": secrets,
Actions: reusable workflows in the repository2103 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
GitHub Actions on g1t, part two: running workflows2104 "matrix": matrix,
2105 "needs": needs,
2106 "strategy": {
2107 "fail-fast": spec.fail_fast,
2108 "job-index": job.ordinal,
2109 "job-total": siblings,
2110 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
2111 },
Fast pages, required checks on the branch, self-hosted runners, honest incidents2112 "runner": runner,
GitHub Actions on g1t, part two: running workflows2113 },
2114 "checkout": {
2115 "repository": run.repo,
2116 "url": clone_url,
2117 "sha": run.sha,
2118 "ref": run.git_ref,
2119 "token": token,
2120 },
2121 "timeoutMinutes": job.timeout_minutes,
2122 "masks": masks,
Merge branch 'worktree-agent-a3abfcce648e87dca'2123 // As the job's log lists them at its start.
2124 "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 R22125 // Whether the job may ask for an OIDC token decides whether it
2126 // is told where to.
2127 "runtime": {
2128 "token": runtime_token,
2129 "idToken": self.oidc_allowed(&run, &job),
2130 },
GitHub Actions on g1t, part two: running workflows2131 })))
2132 }
2133
2134 /// `job_report`: the sandbox telling how the job is going.
2135 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
2136 let job = check!(self.job_for_token(&a).await?);
2137 let report = &a.report;
2138 let at = now();
2139 match report["kind"].as_str().unwrap_or_default() {
2140 "steps" => {
2141 // The list can grow as the job goes (post steps), so steps
2142 // already reported keep where they stand.
2143 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2144 let steps: Vec<Value> = report["steps"]
2145 .as_array()
2146 .map(|names| {
2147 names
2148 .iter()
2149 .enumerate()
2150 .map(|(i, name)| match known.get(i) {
2151 Some(step) if step["status"] != "queued" => step.clone(),
2152 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
2153 })
2154 .collect()
2155 })
2156 .unwrap_or_default();
2157 self.db
2158 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2159 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2160 .run()
2161 .await?;
2162 }
2163 "step" => {
2164 let number = report["number"].as_u64().unwrap_or(0) as usize;
2165 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
2166 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
2167 let status = report["status"].as_str().unwrap_or("in_progress");
2168 step["status"] = json!(status);
2169 if status == "in_progress" {
2170 step["startedAt"] = json!(at);
2171 }
2172 if status == "completed" {
2173 step["finishedAt"] = json!(at);
2174 step["conclusion"] = report["conclusion"].clone();
2175 }
2176 if let Some(name) = report["name"].as_str() {
2177 step["name"] = json!(name);
2178 }
2179 }
2180 self.db
2181 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
2182 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
2183 .run()
2184 .await?;
2185 }
2186 "log" => {
2187 let mut text = report["text"].as_str().unwrap_or_default().to_owned();
2188 if text.len() > MAX_CHUNK_BYTES {
2189 let mut cut = MAX_CHUNK_BYTES;
2190 while !text.is_char_boundary(cut) {
2191 cut -= 1;
2192 }
2193 text.truncate(cut);
2194 }
2195 #[derive(Deserialize)]
2196 struct Size {
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs2197 n: Option<f64>,
2198 seq: Option<f64>,
GitHub Actions on g1t, part two: running workflows2199 }
2200 let size = self
2201 .db
2202 .prepare("SELECT SUM(LENGTH(text)) AS n, MAX(seq) AS seq FROM logs WHERE job_id = ?")
2203 .bind(&[job.id.as_str().into()])?
2204 .first::<Size>(None)
2205 .await?;
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs2206 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 workflows2207 if used < MAX_LOG_BYTES {
2208 if used + text.len() >= MAX_LOG_BYTES {
2209 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
2210 }
2211 self.db
2212 .prepare("INSERT INTO logs (job_id, seq, step, text) VALUES (?, ?, ?, ?)")
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs2213 .bind(&[job.id.as_str().into(), // Numbers go to D1 as f64: a u64 would be a BigInt, which it refuses.
2214 (seq + 1.0).into(), (report["step"].as_u64().unwrap_or(0) as u32).into(), text.into()])?
GitHub Actions on g1t, part two: running workflows2215 .run()
2216 .await?;
2217 }
2218 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2219 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2220 "summary" => {
2221 // `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the
2222 // step's summary, up to 1 MiB a step and 20 steps a job.
2223 let step = report["step"].as_u64().unwrap_or(0) as u32;
2224 let markdown = report["markdown"].as_str().unwrap_or_default();
2225 #[derive(Deserialize)]
2226 struct Held {
2227 steps: u32,
2228 mine: Option<f64>,
2229 }
2230 let held = self
2231 .db
2232 .prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")
2233 .bind(&[step.into(), job.id.as_str().into()])?
2234 .first::<Held>(None)
2235 .await?;
2236 let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));
2237 if summary_fits(steps, mine, markdown.len()) {
2238 self.db
2239 .prepare(
2240 "INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)
2241 ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",
2242 )
2243 .bind(&[job.id.as_str().into(), step.into(), markdown.into()])?
2244 .run()
2245 .await?;
2246 }
2247 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2248 }
GitHub Actions on g1t, part two: running workflows2249 "annotation" => {
2250 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
2251 if annotations.len() < MAX_ANNOTATIONS {
2252 annotations.push(json!({
2253 "level": report["level"].as_str().unwrap_or("notice"),
2254 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
2255 "title": report["title"],
2256 "file": report["file"],
2257 "line": report["line"],
2258 }));
2259 self.db
2260 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
2261 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
2262 .run()
2263 .await?;
2264 }
2265 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2266 // Nothing to tell; the answer says whether to stop.
2267 "ping" => {
2268 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
2269 }
GitHub Actions on g1t, part two: running workflows2270 "done" => {
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2271 // A job told to stop ends cancelled, however its cleanup went.
2272 let conclusion = if job.cancel_requested_at.is_some() {
2273 "cancelled"
2274 } else {
2275 report["conclusion"]
2276 .as_str()
2277 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
2278 .unwrap_or("failure")
2279 };
GitHub Actions on g1t, part two: running workflows2280 let outputs = report["outputs"].as_object().cloned();
2281 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
2282 }
2283 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
2284 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2285 // `cancelled`: the run was cancelled, and the runner should stop
2286 // the step it is on and run only its cleanup steps.
2287 Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))
GitHub Actions on g1t, part two: running workflows2288 }
2289
2290 // --- Every minute ---------------------------------------------------------------
2291
2292 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
2293 let minute = now_ms / 60_000 * 60_000;
Merge platform pause and the hourly usage watcher: staff can pause compute, schedules, indexing or renders for everyone, the watcher emails on a breach and is never blind quietly, and the models proxy holds each run to its cap (billing 0051, integrations 0006)2294 // Staff (or billing's usage watcher) paused scheduled runs across
2295 // g1t: this minute's schedules are skipped, not queued for later.
2296 // Kept 30 seconds in the isolate (g1t_kit::pause).
2297 if g1t_kit::pause::paused(&self.billing, g1t_contracts::billing::PauseLevel::Schedules).await {
2298 worker::console_log!("actions: schedules are paused across g1t; skipped this minute's");
2299 } else if let Err(error) = self.run_schedules(minute).await {
GitHub Actions on g1t, part two: running workflows2300 worker::console_error!("actions: schedules failed: {error}");
2301 }
2302 // Jobs whose sandbox went quiet or ran past their time.
2303 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2304 for job in running {
2305 // Times in g1t's format compare as text.
2306 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
2307 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
2308 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
2309 let over = job.started_at.as_deref().is_some_and(|started| started < before(limit).as_str());
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2310 let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());
2311 if cancel_overdue {
2312 // Cancelled, and still going after its grace period.
2313 self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;
2314 self.advance(&job.run_id).await?;
2315 } else if over {
GitHub Actions on g1t, part two: running workflows2316 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 incidents2317 // A self-hosted runner is told to stop on its next poll.
2318 if job.runner_id.is_none() {
2319 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2320 }
GitHub Actions on g1t, part two: running workflows2321 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2322 } else if silent {
Fast pages, required checks on the branch, self-hosted runners, honest incidents2323 let reason = match &job.runner_name {
2324 Some(name) => format!("The self-hosted runner {name} stopped answering."),
2325 None => "The runner stopped answering.".to_owned(),
2326 };
2327 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
GitHub Actions on g1t, part two: running workflows2328 }
2329 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2330 if let Err(error) = self.sweep_runners(now_ms).await {
2331 worker::console_error!("actions: the runners' sweep failed: {error}");
2332 }
Merge branch 'worktree-agent-a3abfcce648e87dca'2333 // Jobs held at an environment whose wait timer has run out.
2334 if let Err(error) = self.release_gates().await {
2335 worker::console_error!("actions: environments' gates failed: {error}");
2336 }
Fast pages, required checks on the branch, self-hosted runners, honest incidents2337 // Once an hour: the cache's expired entries, and its storage.
2338 if (now_ms / 60_000) % 60 == 7
2339 && let Err(error) = self.sweep_cache(now_ms).await
2340 {
2341 worker::console_error!("actions: the cache's sweep failed: {error}");
2342 }
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R22343 // And artifacts past their time, and the toolkit's abandoned parts.
2344 if (now_ms / 60_000) % 60 == 37 {
2345 if let Err(error) = self.sweep_artifacts(now_ms).await {
2346 worker::console_error!("actions: the artifacts' sweep failed: {error}");
2347 }
2348 if let Err(error) = self.sweep_blob_parts(now_ms).await {
2349 worker::console_error!("actions: the blob parts' sweep failed: {error}");
2350 }
2351 }
GitHub Actions on g1t, part two: running workflows2352 self.start_queued().await
2353 }
2354}
2355
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look2356
2357#[cfg(test)]
2358mod stopping {
2359 use super::stops_runs;
2360 use g1t_contracts::events::Event;
2361 use serde_json::{Value, json};
2362
2363 fn event(kind: &str, data: Value) -> Event {
2364 Event {
2365 id: "evt_1".into(),
2366 kind: kind.into(),
2367 source: "repos".into(),
2368 time: "2026-10-05T00:00:00Z".into(),
2369 repo_id: Some("rep_1".into()),
2370 actor: None,
2371 data,
2372 }
2373 }
2374
2375 #[test]
2376 fn deleting_or_archiving_stops_runs() {
2377 assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));
2378 assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));
2379 assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);
2380 assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);
2381 assert_eq!(stops_runs(&event("git.push", json!({}))), None);
2382 }
2383}
Fast pages, required checks on the branch, self-hosted runners, honest incidents2384
2385#[cfg(test)]
2386mod status_of_needs {
2387 use std::collections::HashMap;
2388
2389 use super::ancestor_failed;
2390
2391 /// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,
2392 /// and a job that needs only the last.
2393 fn graph() -> HashMap<&'static str, Vec<&'static str>> {
2394 HashMap::from([
2395 ("check", vec![]),
2396 ("plan", vec!["check"]),
2397 ("migrate", vec!["plan"]),
2398 ("core", vec!["plan", "migrate"]),
2399 ("edge", vec!["plan", "migrate", "core"]),
2400 ("notify", vec!["edge"]),
2401 ])
2402 }
2403
2404 #[test]
2405 fn a_failure_is_seen_however_far_back() {
2406 let needs = graph();
2407 let failed = |which: &'static str| move |key: &str| key == which;
2408 // check failed; plan, and everything after, was skipped for it.
2409 assert!(ancestor_failed(&needs, "notify", failed("check")));
2410 assert!(ancestor_failed(&needs, "core", failed("check")));
2411 assert!(ancestor_failed(&needs, "edge", failed("core")));
2412 // Nothing before a job failed: a skipped migrate is not a failure.
2413 assert!(!ancestor_failed(&needs, "edge", |_| false));
2414 assert!(!ancestor_failed(&needs, "core", failed("edge")));
2415 assert!(!ancestor_failed(&needs, "check", failed("check")));
2416 }
2417
2418 #[test]
2419 fn cycles_and_unknown_keys_end() {
2420 let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);
2421 assert!(!ancestor_failed(&needs, "a", |_| false));
2422 assert!(!ancestor_failed(&needs, "missing", |_| true));
2423 }
2424}
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb972425
2426#[cfg(test)]
2427mod deployments {
2428 use serde_json::{Map, Value, json};
2429
2430 use super::{JobEnvironment, deployment_outcome, environment_of};
2431
2432 #[test]
2433 fn a_jobs_environment_is_read_for_deployments() {
2434 let contexts: Map<String, Value> = serde_json::from_value(json!({
2435 "github": { "ref_name": "main", "repository": "acme/web" },
2436 "inputs": { "target": "staging" },
2437 "matrix": {},
2438 }))
2439 .unwrap();
2440 let read = |raw: Value| environment_of(&raw, &contexts);
2441 assert_eq!(read(json!({})), None);
2442 assert_eq!(
2443 read(json!({ "environment": "production" })),
2444 Some(JobEnvironment { name: "production".into(), url: None, deploys: true })
2445 );
2446 assert_eq!(
2447 read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),
2448 Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })
2449 );
2450 // Expressions are filled in from the run.
2451 assert_eq!(
2452 read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),
2453 Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })
2454 );
2455 // Secrets only: no deployment.
2456 assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);
2457 // Only http(s) addresses.
2458 assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);
2459 }
2460
2461 #[test]
2462 fn a_runs_outcome_for_an_environment() {
2463 let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());
2464 assert_eq!(of(&["success", "skipped"]), Some("success"));
2465 assert_eq!(of(&["success", "failure"]), Some("failure"));
2466 assert_eq!(of(&["success", "cancelled"]), Some("error"));
2467 assert_eq!(of(&["skipped"]), None);
2468 assert_eq!(deployment_outcome(&[None]), None);
2469 }
2470}
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)2471
2472#[cfg(test)]
2473mod reruns {
2474 use serde_json::{Map, Value, json};
2475
2476 use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, debug_logging, rerun_keys, summary_fits, top_key};
2477
2478 /// build ← test ← deploy, and lint on its own.
2479 fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {
2480 let order = ["build", "lint", "test", "deploy"];
2481 let needs = |key: &str| -> Vec<String> {
2482 match key {
2483 "test" => vec!["build".into()],
2484 "deploy" => vec!["test".into()],
2485 _ => Vec::new(),
2486 }
2487 };
2488 rerun_keys(&order, needs, |key| !failed.contains(&key), which)
2489 }
2490
2491 #[test]
2492 fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {
2493 assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);
2494 assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);
2495 assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);
2496 assert!(keys(Rerun::Failed, &[]).is_empty());
2497 // One job, whatever it came to, and what depends on it.
2498 assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);
2499 assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);
2500 assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);
2501 assert!(keys(Rerun::Job("missing"), &[]).is_empty());
2502 }
2503
2504 #[test]
2505 fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {
2506 assert_eq!(top_key("build/test"), "build");
2507 assert_eq!(top_key("build/inner/test"), "build");
2508 assert_eq!(top_key("lint"), "lint");
2509 }
2510
2511 #[test]
2512 fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {
2513 assert!(summary_fits(0, None, 10));
2514 assert!(!summary_fits(0, None, 0), "nothing to keep");
2515 assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");
2516 assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");
2517 assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));
2518 assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));
2519 assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));
2520 }
2521
2522 #[test]
2523 fn a_debug_rerun_sets_what_github_sets() {
2524 let mut variables = Map::new();
2525 let mut runner = json!({ "name": "g1t", "debug": "" });
2526 debug_logging(&mut variables, &mut runner);
2527 assert_eq!(variables["RUNNER_DEBUG"], "1");
2528 assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");
2529 assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");
2530 assert_eq!(runner["debug"], Value::String("1".into()));
2531 assert_eq!(runner["name"], "g1t");
2532 }
2533}

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