Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/actions/src/plan.rs

1,509 lines68,611 bytesCodeBlame

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};
10use g1t_contracts::actions::{JobCallArgs, RunActionArgs, StartJobArgs, WorkflowRun};
11use g1t_contracts::identity::{CreateAccessTokenArgs, CreatedAccessToken};
12use g1t_contracts::repos::{Repo, RepoPath};
13use g1t_contracts::time::rfc3339;
14use g1t_contracts::{FailureCode, Outcome, new_id};
15use g1t_kit::now_ms;
16use g1t_secrets::{random_hex, same, sha256_hex};
17use serde::Deserialize;
18use serde_json::{Map, Value, json};
19use worker::Result;
20
21use crate::sync::WorkflowRow;
22use crate::{Actions, Count, MAX_TIMEOUT_MINUTES, RUNNING_PER_WORKSPACE, SILENT_MS, SITE, check, fail, optional, repo_path};
23
24/// The most log one job keeps, in bytes; past it, the log says so and stops.
25const MAX_LOG_BYTES: usize = 4 * 1024 * 1024;
26/// The most a single log report may add.
27const MAX_CHUNK_BYTES: usize = 256 * 1024;
28const MAX_ANNOTATIONS: usize = 50;
29
30pub struct NewRun {
31 pub repo: Repo,
32 pub path: String,
33 pub source: String,
34 pub workflow: Workflow,
35 pub info: RunInfo,
36 pub action: Option<String>,
37 pub pull: Option<u32>,
38 pub title: String,
39 pub inputs: Map<String, Value>,
40 pub event_key: String,
41 pub actor_id: Option<String>,
42 pub actor: Option<String>,
43 pub trusted: bool,
44}
45
46#[derive(Clone, Deserialize)]
47pub struct RunRow {
48 pub id: String,
49 pub workflow_id: String,
50 pub repo_id: String,
51 pub repo: String,
52 pub path: String,
53 pub name: String,
54 pub title: String,
55 pub number: u64,
56 pub attempt: u64,
57 pub event: String,
58 pub action: Option<String>,
59 pub git_ref: String,
60 pub sha: String,
61 pub pull: Option<u32>,
62 pub status: String,
63 pub conclusion: Option<String>,
64 pub error: Option<String>,
65 pub actor: Option<String>,
66 pub actor_id: Option<String>,
67 pub source: String,
68 pub info: String,
69 pub inputs: String,
70 pub trusted: u32,
71 pub concurrency_group: Option<String>,
72 pub created_at: String,
73 pub started_at: Option<String>,
74 pub finished_at: Option<String>,
75}
76
77impl RunRow {
78 pub fn info(&self) -> RunInfo {
79 let mut info: RunInfo = serde_json::from_str(&self.info).unwrap_or_default();
80 info.run_id = self.id.clone();
81 info.run_number = self.number;
82 info.run_attempt = self.attempt;
83 info.workflow_path = self.path.clone();
84 if info.workflow.is_empty() {
85 info.workflow = self.name.clone();
86 }
87 info
88 }
89
90 pub fn inputs(&self) -> Map<String, Value> {
91 serde_json::from_str(&self.inputs).unwrap_or_default()
92 }
93
94 pub fn summary(&self) -> WorkflowRun {
95 WorkflowRun {
96 id: self.id.clone(),
97 workflow_id: self.workflow_id.clone(),
98 path: self.path.clone(),
99 name: self.name.clone(),
100 title: self.title.clone(),
101 number: self.number,
102 attempt: self.attempt,
103 event: self.event.clone(),
104 git_ref: self.git_ref.clone(),
105 sha: self.sha.clone(),
106 pull: self.pull,
107 status: self.status.clone(),
108 conclusion: self.conclusion.clone(),
109 error: self.error.clone(),
110 actor: self.actor.clone(),
111 created_at: self.created_at.clone(),
112 started_at: self.started_at.clone(),
113 finished_at: self.finished_at.clone(),
114 }
115 }
116}
117
118#[derive(Clone, Deserialize)]
119pub struct JobRow {
120 pub id: String,
121 pub run_id: String,
122 pub repo_id: String,
123 pub namespace: String,
124 pub key: String,
125 pub ordinal: u32,
126 pub name: String,
127 pub needs: String,
128 pub matrix: Option<String>,
129 pub status: String,
130 pub conclusion: Option<String>,
131 pub steps: String,
132 pub annotations: String,
133 pub outputs: String,
134 pub reason: Option<String>,
135 pub token_hash: Option<String>,
136 pub timeout_minutes: u32,
137 pub continue_on_error: u32,
138 pub max_parallel: Option<u32>,
Actions: reusable workflows in the repository139 /// Set for a job that calls a reusable workflow, and for that
140 /// workflow's jobs (see migration 0002).
141 pub call: Option<String>,
GitHub Actions on g1t, part two: running workflows142 pub seen_at: Option<String>,
143 pub started_at: Option<String>,
144 pub finished_at: Option<String>,
145}
146
147impl JobRow {
148 pub fn needs(&self) -> Vec<String> {
149 serde_json::from_str(&self.needs).unwrap_or_default()
150 }
Actions: reusable workflows in the repository151
152 pub fn call(&self) -> Option<Value> {
153 self.call.as_deref().and_then(|call| serde_json::from_str(call).ok())
154 }
155
156 /// For a job of a called workflow: that workflow, the job's own id in
157 /// it, and the job.
158 pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> {
159 let call = self.call().filter(|call| call["role"] == "callee")?;
160 let called = workflow::parse(call["source"].as_str()?).ok()?;
161 let job = called.jobs.iter().find(|job| call["job"].as_str() == Some(job.id.as_str()))?.clone();
162 Some((called, job, call))
163 }
GitHub Actions on g1t, part two: running workflows164}
165
Actions: reusable workflows in the repository166/// A job to decide on: its key, its definition, and what it needs, as
167/// (name in `needs`, key of the jobs).
168type Unit = (String, workflow::Job, Vec<(String, String)>);
169
170/// How deep reusable workflows may call one another, as on GitHub.
171const MAX_CALL_DEPTH: u64 = 4;
172
GitHub Actions on g1t, part two: running workflows173/// What the jobs of one key came to, for `needs.<key>`.
174fn key_result(rows: &[&JobRow]) -> &'static str {
175 let failed = |row: &&&JobRow| row.conclusion.as_deref() == Some("failure") && row.continue_on_error == 0;
176 if rows.iter().any(|row| failed(&row)) {
177 "failure"
178 } else if rows.iter().any(|row| row.conclusion.as_deref() == Some("cancelled")) {
179 "cancelled"
180 } else if rows.iter().all(|row| row.conclusion.as_deref() == Some("skipped")) {
181 "skipped"
182 } else {
183 "success"
184 }
185}
186
187fn now() -> String {
188 rfc3339(now_ms())
189}
190
191impl Actions {
192 pub async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {
193 self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<RunRow>(None).await
194 }
195
196 pub async fn job_rows(&self, run_id: &str) -> Result<Vec<JobRow>> {
197 self.db
198 .prepare("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid")
199 .bind(&[run_id.into()])?
200 .all()
201 .await?
202 .results::<JobRow>()
203 }
204
205 pub async fn run_summary(&self, id: &str) -> Result<Outcome<WorkflowRun>> {
206 Ok(match self.run_row(id).await? {
207 Some(row) => Outcome::Ok(row.summary()),
208 None => fail(FailureCode::NotFound, "No such run."),
209 })
210 }
211
212 /// The contexts every expression outside a job's steps may use.
213 fn base_contexts(run: &RunRow, vars: &Map<String, Value>, job: &str) -> Map<String, Value> {
214 let mut contexts = Map::new();
215 contexts.insert("github".into(), run.info().context(job, "", run.action.as_deref()));
216 contexts.insert("inputs".into(), Value::Object(run.inputs()));
217 contexts.insert("vars".into(), Value::Object(vars.clone()));
218 contexts.insert("needs".into(), json!({}));
219 contexts.insert("runner".into(), runner_context());
220 contexts
221 }
222
223 /// Makes a run and its jobs, and starts what can start. `None` when the
224 /// event already started this workflow.
225 pub async fn create_run(&self, new: NewRun) -> Result<Option<String>> {
226 let workflow_row = self.workflow_row(&new.repo, &new.path, &new.workflow.display_name(&new.path), &new.source).await?;
227 let numbered = self
228 .db
229 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
230 .bind(&[workflow_row.id.as_str().into()])?
231 .first::<Count>(None)
232 .await?
233 .map_or(1, |count| count.n);
234 let id = new_id("run", now_ms());
235 let mut info = new.info.clone();
236 info.workflow = new.workflow.display_name(&new.path);
237 info.workflow_path = new.path.clone();
238 info.run_id = id.clone();
239 info.run_number = u64::from(numbered);
Secrets and variables: one list, rows per environment, for workflows and deployments240 let vars = self
241 .variables_for(&new.repo.id, &format!("{}/{}", new.repo.namespace, new.repo.name), None, new.trusted)
242 .await?;
GitHub Actions on g1t, part two: running workflows243
244 // run-name and the concurrency group read github, inputs and vars.
245 let mut contexts = Map::new();
246 contexts.insert("github".into(), info.context("", "", new.action.as_deref()));
247 contexts.insert("inputs".into(), Value::Object(new.inputs.clone()));
248 contexts.insert("vars".into(), Value::Object(vars));
249 let scope = Scope {
250 contexts: &contexts,
251 status: Status::Success,
252 hash_files: None,
253 };
254 let title = new
255 .workflow
256 .run_name
257 .as_deref()
258 .and_then(|run_name| expr::interpolate(run_name, &scope).ok())
259 .filter(|title| !title.trim().is_empty())
260 .unwrap_or(new.title.clone());
261 let group = new.workflow.concurrency.as_ref().and_then(|c| expr::interpolate(&c.group, &scope).ok());
262 let cancel_in_progress = new
263 .workflow
264 .concurrency
265 .as_ref()
266 .and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())
267 .is_some_and(|value| expr::truthy(&value));
268
269 let inserted = self
270 .db
271 .prepare(
272 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, action, git_ref, sha,
273 pull, status, actor, actor_id, source, info, inputs, trusted, concurrency_group, event_key, created_at)
274 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
275 )
276 .bind(&[
277 id.as_str().into(),
278 workflow_row.id.as_str().into(),
279 new.repo.id.as_str().into(),
280 format!("{}/{}", new.repo.namespace, new.repo.name).into(),
281 new.path.as_str().into(),
282 info.workflow.as_str().into(),
283 title.as_str().into(),
284 numbered.into(),
285 info.event_name.as_str().into(),
286 optional(new.action.as_deref()),
287 info.git_ref.as_str().into(),
288 info.sha.as_str().into(),
289 new.pull.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
290 optional(new.actor.as_deref()),
291 optional(new.actor_id.as_deref()),
292 new.source.as_str().into(),
293 serde_json::to_string(&info)?.into(),
294 serde_json::to_string(&new.inputs)?.into(),
295 u32::from(new.trusted).into(),
296 optional(group.as_deref()),
297 new.event_key.as_str().into(),
298 now().into(),
299 ])?
300 .first::<Value>(None)
301 .await?;
302 if inserted.is_none() {
303 return Ok(None);
304 }
305 if let Some(run) = self.run_row(&id).await? {
306 self.report_pending(&run).await?;
307 }
308
309 // Every job, waiting; each is expanded when the jobs it needs are done.
310 let mut statements = Vec::new();
311 for job in &new.workflow.jobs {
312 statements.push(
313 self.db
314 .prepare(
315 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting')",
316 )
317 .bind(&[
318 new_id("job", now_ms()).into(),
319 id.as_str().into(),
320 new.repo.id.as_str().into(),
321 new.repo.namespace.as_str().into(),
322 job.id.as_str().into(),
323 job.name.clone().filter(|n| !expr::has_expression(n)).unwrap_or(job.id.clone()).into(),
324 serde_json::to_string(&job.needs)?.into(),
325 ])?,
326 );
327 }
328 self.db.batch(statements).await?;
329
330 // One run at a time per concurrency group.
331 if let Some(group) = &group {
332 let others = self
333 .db
334 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND id != ? AND status != 'completed' ORDER BY id")
335 .bind(&[new.repo.id.as_str().into(), group.as_str().into(), id.as_str().into()])?
336 .all()
337 .await?
338 .results::<RunRow>()?;
339 for other in &others {
340 if cancel_in_progress || other.status == "pending" {
341 // A newer run replaces a waiting one, as on GitHub.
342 self.cancel_run(other, "A newer run in the same concurrency group replaced it.").await?;
343 }
344 }
345 if !cancel_in_progress && others.iter().any(|other| other.status != "pending") {
346 self.db
347 .prepare("UPDATE runs SET status = 'pending' WHERE id = ?")
348 .bind(&[id.as_str().into()])?
349 .run()
350 .await?;
351 return Ok(Some(id));
352 }
353 }
354 self.advance(&id).await?;
355 Ok(Some(id))
356 }
357
358 /// A run that could not start, such as for a workflow file that does not read.
359 #[allow(clippy::too_many_arguments)]
360 pub async fn record_failed_run(
361 &self,
362 row: &WorkflowRow,
363 git_ref: &str,
364 sha: &str,
365 event_key: &str,
366 actor_id: Option<&str>,
367 actor: &str,
368 problem: &str,
369 ) -> Result<()> {
370 let numbered = self
371 .db
372 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
373 .bind(&[row.id.as_str().into()])?
374 .first::<Count>(None)
375 .await?
376 .map_or(1, |count| count.n);
377 let at = now();
378 self.db
379 .prepare(
380 "INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, git_ref, sha, status,
381 conclusion, error, actor, actor_id, source, info, event_key, created_at, finished_at)
382 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'push', ?, ?, 'completed', 'failure', ?, ?, ?, ?, '{}', ?, ?, ?)",
383 )
384 .bind(&[
385 new_id("run", now_ms()).into(),
386 row.id.as_str().into(),
387 row.repo_id.as_str().into(),
388 row.repo.as_str().into(),
389 row.path.as_str().into(),
390 row.path.as_str().into(),
391 "Invalid workflow file".into(),
392 numbered.into(),
393 git_ref.into(),
394 sha.into(),
395 problem.into(),
396 actor.into(),
397 optional(actor_id),
398 row.source.as_str().into(),
399 event_key.into(),
400 at.as_str().into(),
401 at.as_str().into(),
402 ])?
403 .run()
404 .await?;
405 Ok(())
406 }
407
408 /// Moves a run along: jobs whose needs are done are decided on, and
409 /// jobs that can start are started.
410 pub async fn advance(&self, run_id: &str) -> Result<()> {
411 // Each pass may finish jobs (skipped ones), which may free others.
412 for _ in 0..20 {
413 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
414 if run.status == "completed" || run.status == "pending" {
415 return Ok(());
416 }
417 let workflow = match workflow::parse(&run.source) {
418 Ok(workflow) => workflow,
419 Err(problem) => {
420 self.finish_run(&run, Some(&problem)).await?;
421 return Ok(());
422 }
423 };
424 let jobs = self.job_rows(run_id).await?;
425 let mut changed = false;
Actions: reusable workflows in the repository426 // What to decide on: the workflow's jobs, and the jobs of the
427 // workflows they call, each with its needs as (name, key).
428 let mut units: Vec<Unit> = workflow
429 .jobs
430 .iter()
431 .map(|job| (job.id.clone(), job.clone(), job.needs.iter().map(|n| (n.clone(), n.clone())).collect()))
432 .collect();
433 let mut seen = std::collections::HashSet::new();
434 for row in &jobs {
435 if !seen.insert(row.key.clone()) {
436 continue;
437 }
438 if let Some((_, job, call)) = row.callee() {
439 let parent = call["parent"].as_str().unwrap_or_default().to_owned();
440 let needs = job.needs.iter().map(|n| (n.clone(), format!("{parent}/{n}"))).collect();
441 units.push((row.key.clone(), job, needs));
442 }
443 }
444 for (key, job, needs) in &units {
445 let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect();
GitHub Actions on g1t, part two: running workflows446 if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting") {
447 continue;
448 }
449 let needed: Vec<(&String, Vec<&JobRow>)> =
Actions: reusable workflows in the repository450 needs.iter().map(|(name, need)| (name, jobs.iter().filter(|row| &row.key == need).collect())).collect();
GitHub Actions on g1t, part two: running workflows451 if !needed.iter().all(|(_, rows)| rows.iter().all(|row| row.status == "completed")) {
452 continue;
453 }
454 self.decide(&run, job, rows[0], &needed).await?;
455 changed = true;
456 }
Actions: reusable workflows in the repository457 // A job that called a workflow finishes with that workflow's jobs.
458 for row in jobs.iter().filter(|row| row.status == "calling") {
459 let children: Vec<&JobRow> = jobs
460 .iter()
461 .filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(row.key.as_str())))
462 .collect();
463 if !children.is_empty() && children.iter().all(|child| child.status == "completed") {
464 self.finish_call(row, &children).await?;
465 changed = true;
466 }
467 }
GitHub Actions on g1t, part two: running workflows468 if !changed {
469 break;
470 }
471 }
472 self.start_queued().await?;
473 self.finish_if_done(run_id).await
474 }
475
476 /// Decides on one job whose needs are done: skip it, fail it, or expand
477 /// it into its matrix and queue it.
478 async fn decide(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, needed: &[(&String, Vec<&JobRow>)]) -> Result<()> {
Secrets and variables: one list, rows per environment, for workflows and deployments479 let vars = self.variables_for(&run.repo_id, &run.repo, None, run.trusted != 0).await?;
GitHub Actions on g1t, part two: running workflows480 let mut contexts = Self::base_contexts(run, &vars, &job.id);
Actions: reusable workflows in the repository481 // A called workflow's jobs read the inputs they were called with.
482 let call = row.call();
483 let parent = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));
484 if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") {
485 contexts.insert("inputs".into(), call["inputs"].clone());
486 }
GitHub Actions on g1t, part two: running workflows487 let mut needs = Map::new();
488 let mut status = if run.conclusion.as_deref() == Some("cancelled") { Status::Cancelled } else { Status::Success };
489 for (key, rows) in needed {
490 let result = key_result(rows);
491 let mut outputs = Map::new();
492 for row in rows {
493 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
494 outputs.extend(more);
495 }
496 }
497 if result != "success" && status == Status::Success {
498 status = Status::Failure;
499 }
500 needs.insert((*key).clone(), json!({ "result": result, "outputs": outputs }));
501 }
502 contexts.insert("needs".into(), Value::Object(needs));
503 let scope = Scope {
504 contexts: &contexts,
505 status,
506 hash_files: None,
507 };
508 let condition = job.condition.as_deref().unwrap_or_default();
509 match expr::condition(condition, &scope) {
510 Ok(true) => {}
511 Ok(false) => return self.skip_job(row, None).await,
512 Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await,
513 }
Actions: reusable workflows in the repository514 if let Some(uses) = &job.uses {
515 return self.call_workflow(run, job, row, uses, &scope).await;
GitHub Actions on g1t, part two: running workflows516 }
517
518 // Its matrix, which may come from a needed job's outputs.
519 let combinations = match &job.matrix {
520 None => vec![Map::new()],
521 Some(matrix) => {
522 let value = match expr::interpolate_value(matrix, &scope) {
523 Ok(value) => value,
524 Err(problem) => return self.fail_job(row, &format!("Its matrix does not read: {problem}")).await,
525 };
526 match matrix::expand(&value) {
527 Ok(combinations) if !combinations.is_empty() => combinations,
528 Ok(_) => return self.fail_job(row, "Its matrix makes no jobs.").await,
529 Err(problem) => return self.fail_job(row, &problem).await,
530 }
531 }
532 };
533 let total = combinations.len();
534 let raw = job.raw.as_object().cloned().unwrap_or_default();
535 let mut statements = Vec::new();
536 for (index, combination) in combinations.iter().enumerate() {
537 let mut contexts = contexts.clone();
538 contexts.insert("matrix".into(), Value::Object(combination.clone()));
539 contexts.insert(
540 "strategy".into(),
541 json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),
542 );
543 let scope = Scope {
544 contexts: &contexts,
545 status: Status::Success,
546 hash_files: None,
547 };
548 let base_name = job.name.clone().unwrap_or(job.id.clone());
Actions: reusable workflows in the repository549 // A called workflow's job is shown under the job that called it.
550 let base_name = match &parent {
551 Some(parent) => format!("{} / {base_name}", parent.replace('/', " / ")),
552 None => base_name,
553 };
GitHub Actions on g1t, part two: running workflows554 let name = if expr::has_expression(&base_name) {
555 expr::interpolate(&base_name, &scope).unwrap_or(base_name)
556 } else if job.matrix.is_some() {
557 matrix::job_name(&base_name, combination)
558 } else {
559 base_name
560 };
561 let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null);
562 let labels = match &runs_on {
563 Value::String(label) => vec![label.clone()],
564 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
565 Value::Object(spec) => spec.get("labels").map(|l| match l {
566 Value::Array(labels) => labels.iter().map(expr::to_text).collect(),
567 other => vec![expr::to_text(other)],
568 }).unwrap_or_default(),
569 _ => Vec::new(),
570 };
571 let reason = labels
572 .iter()
573 .find(|label| {
574 let lower = label.to_ascii_lowercase();
575 lower.contains("windows") || lower.contains("macos")
576 })
577 .map(|label| format!("`runs-on: {label}`: g1t runs jobs on Linux only."));
578 let timeout = raw
579 .get("timeout-minutes")
580 .and_then(|value| expr::interpolate_value(value, &scope).ok())
581 .and_then(|value| value.as_f64().or_else(|| expr::to_text(&value).parse().ok()))
582 .map_or(MAX_TIMEOUT_MINUTES, |minutes| (minutes.ceil() as u32).clamp(1, MAX_TIMEOUT_MINUTES));
583 let continue_on_error = raw
584 .get("continue-on-error")
585 .and_then(|value| expr::interpolate_value(value, &scope).ok())
586 .is_some_and(|value| expr::truthy(&value));
587 let (status, conclusion, finished) = match &reason {
588 Some(_) => ("completed", Some("failure"), Some(now())),
589 None => ("queued", None, None),
590 };
591 let values: Vec<worker::wasm_bindgen::JsValue> = vec![
592 name.into(),
593 serde_json::to_string(combination)?.into(),
594 status.into(),
595 optional(conclusion),
596 optional(reason.as_deref()),
597 timeout.into(),
598 u32::from(continue_on_error).into(),
599 job.max_parallel.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
600 optional(finished.as_deref()),
601 ];
602 if index == 0 {
603 let mut bound = values;
604 bound.push(row.id.as_str().into());
605 statements.push(
606 self.db
607 .prepare(
608 "UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, timeout_minutes = ?,
609 continue_on_error = ?, max_parallel = ?, finished_at = ? WHERE id = ?",
610 )
611 .bind(&bound)?,
612 );
613 } else {
614 let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![
615 new_id("job", now_ms()).into(),
616 row.run_id.as_str().into(),
617 row.repo_id.as_str().into(),
618 row.namespace.as_str().into(),
619 row.key.as_str().into(),
620 (index as u32).into(),
621 row.needs.as_str().into(),
Actions: reusable workflows in the repository622 optional(row.call.as_deref()),
GitHub Actions on g1t, part two: running workflows623 ];
624 bound.extend(values);
625 statements.push(
626 self.db
627 .prepare(
Actions: reusable workflows in the repository628 "INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, call, name, matrix, status, conclusion, reason,
GitHub Actions on g1t, part two: running workflows629 timeout_minutes, continue_on_error, max_parallel, finished_at)
Actions: reusable workflows in the repository630 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
GitHub Actions on g1t, part two: running workflows631 )
632 .bind(&bound)?,
633 );
634 }
635 }
636 self.db.batch(statements).await?;
637 Ok(())
638 }
639
Actions: reusable workflows in the repository640 /// A job that calls a reusable workflow in the repository: that
641 /// workflow's jobs join the run under it, with the inputs it passes.
642 async fn call_workflow(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, uses: &str, scope: &Scope<'_>) -> Result<()> {
643 let Some(local) = uses.strip_prefix("./") else {
644 return self
645 .fail_job(row, "Reusable workflows from other repositories are not called on g1t yet; ones in this repository (`./.g1t/workflows/…`) are.")
646 .await;
647 };
648 let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;
649 if depth > MAX_CALL_DEPTH {
650 return self.fail_job(row, &format!("Reusable workflows call each other more than {MAX_CALL_DEPTH} deep.")).await;
651 }
652 let local = local.split('@').next().unwrap_or(local).to_owned();
653 let path = repo_path(&run.repo);
654 let Some(ws) = self.workspace_actor(&path.namespace).await? else {
655 return self.fail_job(row, "The workspace is gone.").await;
656 };
657 // A repository moved from GitHub keeps saying `.github/…`.
658 let mut found = self.read_file(&path, &ws, &run.sha, &local).await?.map(|text| (local.clone(), text));
659 if found.is_none()
660 && let Some(rest) = local.strip_prefix(".github/")
661 {
662 let moved = format!(".g1t/{rest}");
663 found = self.read_file(&path, &ws, &run.sha, &moved).await?.map(|text| (moved, text));
664 }
665 let Some((file, source)) = found else {
666 return self.fail_job(row, &format!("`{uses}` is not in the repository at this commit.")).await;
667 };
668 let called = match workflow::parse(&source) {
669 Ok(called) => called,
670 Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,
671 };
672 let Some(trigger) = called.trigger("workflow_call") else {
673 return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;
674 };
675 // Inputs: what the caller passes, else the called workflow's defaults.
676 let given = match job.raw.get("with") {
677 Some(with) => match expr::interpolate_value(with, scope) {
678 Ok(Value::Object(given)) => given,
679 Ok(_) => Map::new(),
680 Err(problem) => return self.fail_job(row, &format!("Its `with` does not read: {problem}")).await,
681 },
682 None => Map::new(),
683 };
684 let mut inputs = Map::new();
685 for (name, spec) in &trigger.inputs {
686 let value = given.get(name).cloned().or_else(|| spec.get("default").cloned()).unwrap_or(Value::Null);
687 if value.is_null() && spec.get("required").and_then(Value::as_bool) == Some(true) {
688 return self.fail_job(row, &format!("`{file}` needs the input `{name}`.")).await;
689 }
690 inputs.insert(name.clone(), value);
691 }
692 for (name, value) in given {
693 inputs.entry(name).or_insert(value);
694 }
695 let mut statements = Vec::new();
696 for called_job in &called.jobs {
697 let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{}/{n}", row.key)).collect();
698 let call = json!({
699 "role": "callee", "parent": row.key, "job": called_job.id, "path": file,
700 "source": source, "inputs": inputs, "depth": depth,
701 });
702 statements.push(
703 self.db
704 .prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")
705 .bind(&[
706 new_id("job", now_ms()).into(),
707 row.run_id.as_str().into(),
708 row.repo_id.as_str().into(),
709 row.namespace.as_str().into(),
710 format!("{}/{}", row.key, called_job.id).into(),
711 format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),
712 serde_json::to_string(&needs)?.into(),
713 serde_json::to_string(&call)?.into(),
714 ])?,
715 );
716 }
717 statements.push(
718 self.db
719 .prepare("UPDATE jobs SET status = 'calling', call = ?, reason = ?, started_at = ? WHERE id = ?")
720 .bind(&[
721 serde_json::to_string(&json!({ "role": "caller", "path": file, "source": source }))?.into(),
722 format!("Calls `{file}`.").into(),
723 now().into(),
724 row.id.as_str().into(),
725 ])?,
726 );
727 self.db.batch(statements).await?;
728 Ok(())
729 }
730
731 /// A job that called a workflow, finished with its jobs: their result,
732 /// and the outputs the workflow declares.
733 async fn finish_call(&self, row: &JobRow, children: &[&JobRow]) -> Result<()> {
734 let call = row.call().unwrap_or_default();
735 let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());
736 let mut jobs_context = Map::new();
737 let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();
738 for child in children {
739 by_key.entry(child.key.clone()).or_default().push(child);
740 }
741 for (key, rows) in &by_key {
742 let mut outputs = Map::new();
743 for child in rows {
744 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {
745 outputs.extend(more);
746 }
747 }
748 let id = key.rsplit('/').next().unwrap_or(key);
749 jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));
750 }
751 let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));
752 let mut contexts = Map::new();
753 contexts.insert("jobs".into(), Value::Object(jobs_context));
754 contexts.insert("inputs".into(), inputs);
755 let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };
756 let mut outputs = Map::new();
757 if let Some(Value::Object(declared)) = called.as_ref().map(|w| {
758 let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);
759 on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)
760 }) {
761 for (name, spec) in declared {
762 if let Some(value) = spec.get("value") {
763 let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);
764 outputs.insert(name, Value::String(expr::to_text(&value)));
765 }
766 }
767 }
768 let conclusion = key_result(children);
769 self.db
770 .prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling'")
771 .bind(&[conclusion.into(), serde_json::to_string(&outputs)?.into(), now().into(), row.id.as_str().into()])?
772 .run()
773 .await?;
774 Ok(())
775 }
776
777 /// A file's text at a commit, if it is there.
778 async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {
779 let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(
780 &self.repos,
781 "blob",
782 &g1t_contracts::repos::BlobArgs {
783 path: path.clone(),
784 viewer: Some(ws.clone()),
785 git_ref: sha.to_owned(),
786 file_path: file.to_owned(),
787 },
788 )
789 .await?;
790 Ok(match blob {
791 Outcome::Ok(view) => view.text,
792 Outcome::Fail(_) => None,
793 })
794 }
795
GitHub Actions on g1t, part two: running workflows796 async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {
797 self.db
798 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")
799 .bind(&[optional(reason), now().into(), row.id.as_str().into()])?
800 .run()
801 .await?;
802 Ok(())
803 }
804
805 async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {
806 self.db
807 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")
808 .bind(&[reason.into(), now().into(), row.id.as_str().into()])?
809 .run()
810 .await?;
811 Ok(())
812 }
813
814 /// Starts queued jobs, oldest first, while their workspace has room.
815 pub async fn start_queued(&self) -> Result<()> {
816 let queued = self
817 .db
818 .prepare("SELECT * FROM jobs WHERE status = 'queued' ORDER BY rowid LIMIT 50")
819 .all()
820 .await?
821 .results::<JobRow>()?;
822 let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
823 for job in queued {
824 let in_workspace = match running.get(&job.namespace) {
825 Some(n) => *n,
826 None => {
827 let n = self
828 .db
829 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND namespace = ?")
830 .bind(&[job.namespace.as_str().into()])?
831 .first::<Count>(None)
832 .await?
833 .map_or(0, |count| count.n);
834 running.insert(job.namespace.clone(), n);
835 n
836 }
837 };
838 if in_workspace >= RUNNING_PER_WORKSPACE {
839 continue;
840 }
841 if let Some(max) = job.max_parallel {
842 let siblings = self
843 .db
844 .prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")
845 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
846 .first::<Count>(None)
847 .await?
848 .map_or(0, |count| count.n);
849 if siblings >= max {
850 continue;
851 }
852 }
853 let token = random_hex(24);
854 let at = now();
855 let claimed = self
856 .db
857 .prepare(
858 "UPDATE jobs SET status = 'in_progress', token_hash = ?, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
859 )
860 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
861 .first::<Value>(None)
862 .await?;
863 if claimed.is_none() {
864 continue;
865 }
866 running.insert(job.namespace.clone(), in_workspace + 1);
867 self.db
868 .prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")
869 .bind(&[at.as_str().into(), job.run_id.as_str().into()])?
870 .run()
871 .await?;
872 let run = self.run_row(&job.run_id).await?;
873 let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {
874 namespace: job.namespace.clone(),
875 name: String::new(),
876 });
877 let started: Outcome<Value> = g1t_kit::call(
878 &self.runner,
879 "start_actions_job",
880 &StartJobArgs {
881 job: job.id.clone(),
882 token,
883 repo,
884 timeout_minutes: job.timeout_minutes,
885 },
886 )
887 .await
888 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
889 if let Outcome::Fail(refused) = started {
890 Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;
891 }
892 }
893 Ok(())
894 }
895
896 /// Finishes a job and moves its run along.
897 pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {
898 let finished = self
899 .db
900 .prepare(
901 "UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),
902 finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",
903 )
904 .bind(&[
905 conclusion.into(),
906 optional(reason),
907 outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
908 now().into(),
909 job_id.into(),
910 ])?
911 .first::<JobRow>(None)
912 .await?;
913 let Some(job) = finished else { return Ok(()) };
914 // Steps still marked as going are not going any more.
915 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
916 let mut touched = false;
917 for step in steps.iter_mut() {
918 if step["status"] != "completed" {
919 let was_running = step["status"] == "in_progress";
920 step["status"] = json!("completed");
921 step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });
922 touched = true;
923 }
924 }
925 if touched {
926 self.db
927 .prepare("UPDATE jobs SET steps = ? WHERE id = ?")
928 .bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?
929 .run()
930 .await?;
931 }
932 // fail-fast: one failed combination stops the rest of its matrix.
933 if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {
934 let run = self.run_row(&job.run_id).await?;
935 let fail_fast = run
936 .as_ref()
937 .and_then(|run| workflow::parse(&run.source).ok())
938 .and_then(|workflow| workflow.jobs.into_iter().find(|j| j.id == job.key))
939 .is_none_or(|j| j.fail_fast);
940 if fail_fast {
941 let siblings = self
942 .db
943 .prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")
944 .bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?
945 .all()
946 .await?
947 .results::<JobRow>()?;
948 for sibling in siblings {
949 self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;
950 }
951 }
952 }
953 Box::pin(self.advance(&job.run_id)).await
954 }
955
956 /// Cancels a job, stopping its sandbox if it has one.
957 async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {
958 if job.status == "in_progress" {
959 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
960 }
961 self.db
962 .prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL WHERE id = ? AND status != 'completed'")
963 .bind(&[reason.into(), now().into(), job.id.as_str().into()])?
964 .run()
965 .await?;
966 Ok(())
967 }
968
969 /// Finishes the run when every job has.
970 async fn finish_if_done(&self, run_id: &str) -> Result<()> {
971 let Some(run) = self.run_row(run_id).await? else { return Ok(()) };
972 if run.status == "completed" || run.status == "pending" {
973 return Ok(());
974 }
975 let jobs = self.job_rows(run_id).await?;
976 if !jobs.iter().all(|job| job.status == "completed") {
977 return Ok(());
978 }
979 self.finish_run(&run, None).await
980 }
981
982 async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {
983 let jobs = self.job_rows(&run.id).await?;
984 let rows: Vec<&JobRow> = jobs.iter().collect();
985 let conclusion = if error.is_some() {
986 "failure"
987 } else if run.conclusion.as_deref() == Some("cancelled") {
988 "cancelled"
989 } else if rows.is_empty() {
990 "skipped"
991 } else {
992 key_result(&rows)
993 };
994 let done = self
995 .db
996 .prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")
997 .bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?
998 .first::<Value>(None)
999 .await?;
1000 if done.is_none() {
1001 return Ok(());
1002 }
1003 self.report_status(run, conclusion).await?;
Actions: workflow_run, workflow.completed, artifacts on the run page, Node 241004 let published: Result<()> = g1t_kit::call(
1005 &self.events,
1006 "publish",
1007 &g1t_contracts::events::Publish {
1008 events: vec![g1t_contracts::events::NewEvent {
1009 kind: "workflow.completed",
1010 source: "actions",
1011 repo_id: Some(run.repo_id.clone()),
1012 actor: run.actor_id.clone(),
1013 data: g1t_contracts::events::WorkflowEvent {
1014 run_id: run.id.clone(),
1015 repo_id: run.repo_id.clone(),
1016 workflow: run.name.clone(),
1017 path: run.path.clone(),
1018 number: run.number,
1019 event: run.event.clone(),
1020 conclusion: conclusion.to_owned(),
1021 git_ref: run.git_ref.clone(),
1022 sha: run.sha.clone(),
1023 pull: run.pull,
1024 },
1025 }],
1026 },
1027 )
1028 .await;
1029 if let Err(error) = published {
1030 worker::console_error!("actions: could not publish workflow.completed: {error}");
1031 }
GitHub Actions on g1t, part two: running workflows1032 // The next run waiting in its concurrency group.
1033 if let Some(group) = &run.concurrency_group {
1034 let next = self
1035 .db
1036 .prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")
1037 .bind(&[run.repo_id.as_str().into(), group.as_str().into()])?
1038 .first::<RunRow>(None)
1039 .await?;
1040 if let Some(next) = next {
1041 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;
1042 Box::pin(self.advance(&next.id)).await?;
1043 }
1044 }
1045 Ok(())
1046 }
1047
1048 /// Tells the pull request (or commit) how the run went, as a status.
1049 async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {
1050 let state = match conclusion {
1051 "success" | "skipped" => "success",
1052 "cancelled" => "error",
1053 _ => "failure",
1054 };
1055 let _: Result<Value> = g1t_kit::call(
1056 &self.work,
1057 "set_commit_status",
1058 &json!({
1059 "repoId": run.repo_id,
1060 "sha": run.sha,
1061 "context": format!("{} / {}", run.name, run.event),
1062 "state": state,
1063 "description": format!("{} {}", run.name, match conclusion {
1064 "success" => "passed",
1065 "skipped" => "was skipped",
1066 "cancelled" => "was cancelled",
1067 _ => "failed",
1068 }),
1069 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1070 }),
1071 )
1072 .await;
1073 Ok(())
1074 }
1075
1076 /// Tells the pull request a run has started on its head.
1077 pub async fn report_pending(&self, run: &RunRow) -> Result<()> {
1078 let _: Result<Value> = g1t_kit::call(
1079 &self.work,
1080 "set_commit_status",
1081 &json!({
1082 "repoId": run.repo_id,
1083 "sha": run.sha,
1084 "context": format!("{} / {}", run.name, run.event),
1085 "state": "pending",
1086 "description": format!("{} is running", run.name),
1087 "targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),
1088 }),
1089 )
1090 .await;
1091 Ok(())
1092 }
1093
1094 /// Cancels a run: its waiting and queued jobs, and stops its running ones.
1095 pub async fn cancel_run(&self, run: &RunRow, reason: &str) -> Result<()> {
1096 self.db
1097 .prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")
1098 .bind(&[run.id.as_str().into()])?
1099 .run()
1100 .await?;
1101 for job in self.job_rows(&run.id).await?.iter().filter(|job| job.status != "completed") {
1102 self.stop_job(job, reason).await?;
1103 }
1104 if run.status == "pending" {
1105 self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;
1106 }
1107 self.advance(&run.id).await
1108 }
1109
1110 pub async fn cancel(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
1111 if let Some(Outcome::Fail(refused)) = Self::member(&a.actor, &a.repo) {
1112 return Ok(Outcome::Fail(refused));
1113 }
1114 let run = check!(self.run_in(&a.repo, &a.id).await?);
1115 if run.status == "completed" {
1116 return Ok(fail(FailureCode::Conflict, "The run has already finished."));
1117 }
1118 self.cancel_run(&run, &format!("{} cancelled the run.", a.actor.username)).await?;
1119 self.run_summary(&run.id).await
1120 }
1121
1122 /// Runs again: every job, or with `failed_only` those that did not
1123 /// succeed and the jobs that need them.
1124 pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {
1125 if let Some(Outcome::Fail(refused)) = Self::member(&a.actor, &a.repo) {
1126 return Ok(Outcome::Fail(refused));
1127 }
1128 let run = check!(self.run_in(&a.repo, &a.id).await?);
1129 if run.status != "completed" {
1130 return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));
1131 }
1132 if run.error.is_some() {
1133 return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));
1134 }
1135 let jobs = self.job_rows(&run.id).await?;
1136 let workflow = workflow::parse(&run.source).ok();
1137 // Which keys run again: failed ones and, transitively, those needing them.
1138 let mut again: Vec<String> = Vec::new();
1139 for key in workflow.as_ref().map(|w| w.job_order()).unwrap_or_default() {
1140 let rows: Vec<&JobRow> = jobs.iter().filter(|j| j.key == key).collect();
1141 let failed = rows.iter().any(|row| row.conclusion.as_deref() != Some("success"));
1142 let needs_again = rows.first().is_some_and(|row| row.needs().iter().any(|need| again.contains(need)));
1143 if !a.failed_only || failed || needs_again {
1144 again.push(key.to_owned());
1145 }
1146 }
1147 if again.is_empty() {
1148 return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));
1149 }
1150 let mut statements = Vec::new();
1151 for key in &again {
1152 statements.push(self.db.prepare("DELETE FROM logs WHERE job_id IN (SELECT id FROM jobs WHERE run_id = ? AND key = ?)").bind(&[run.id.as_str().into(), key.as_str().into()])?);
1153 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 repository1154 // The jobs of a workflow it called are made again when it calls it again.
GitHub Actions on g1t, part two: running workflows1155 statements.push(
1156 self.db
Actions: reusable workflows in the repository1157 .prepare("DELETE FROM logs WHERE job_id IN (SELECT id FROM jobs WHERE run_id = ? AND key LIKE ?)")
1158 .bind(&[run.id.as_str().into(), format!("{key}/%").into()])?,
1159 );
1160 statements.push(
1161 self.db
1162 .prepare("DELETE FROM jobs WHERE run_id = ? AND key LIKE ?")
1163 .bind(&[run.id.as_str().into(), format!("{key}/%").into()])?,
1164 );
1165 statements.push(
1166 self.db
GitHub Actions on g1t, part two: running workflows1167 .prepare(
1168 "UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,
Actions: reusable workflows in the repository1169 matrix = NULL, call = NULL, token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL WHERE run_id = ? AND key = ?",
GitHub Actions on g1t, part two: running workflows1170 )
1171 .bind(&[run.id.as_str().into(), key.as_str().into()])?,
1172 );
1173 }
1174 statements.push(
1175 self.db
1176 .prepare("UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL WHERE id = ?")
1177 .bind(&[run.id.as_str().into()])?,
1178 );
1179 self.db.batch(statements).await?;
1180 if let Some(run) = self.run_row(&run.id).await? {
1181 self.report_pending(&run).await?;
1182 }
1183 self.advance(&run.id).await?;
1184 self.run_summary(&run.id).await
1185 }
1186
1187 pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {
1188 let row = self
1189 .db
1190 .prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")
1191 .bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?
1192 .first::<RunRow>(None)
1193 .await?;
1194 Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))
1195 }
1196
1197 // --- The sandbox's side -----------------------------------------------------
1198
1199 async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {
1200 let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;
1201 Ok(match job {
1202 Some(job) if job.status == "in_progress" && job.token_hash.as_deref().is_some_and(|hash| same(hash, &sha256_hex(&a.token))) => {
1203 Outcome::Ok(job)
1204 }
1205 _ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),
1206 })
1207 }
1208
A repository has its own sidebar, as settings do1209 /// `job_auth`: which run and repository a running job's token is for,
1210 /// so the API can keep its artifacts and cache.
1211 pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1212 let job = check!(self.job_for_token(&a).await?);
1213 Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))
1214 }
1215
GitHub Actions on g1t, part two: running workflows1216 /// `job_spec`: everything the sandbox needs to run the job.
1217 pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1218 let job = check!(self.job_for_token(&a).await?);
1219 let Some(run) = self.run_row(&job.run_id).await? else {
1220 return Ok(fail(FailureCode::NotFound, "No such run."));
1221 };
Actions: reusable workflows in the repository1222 let Ok(caller) = workflow::parse(&run.source) else {
GitHub Actions on g1t, part two: running workflows1223 return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));
1224 };
Actions: reusable workflows in the repository1225 // A called workflow's job runs as that workflow defines it.
1226 let callee = job.callee();
1227 let (workflow, spec, call_inputs) = match callee {
1228 Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),
1229 None => match caller.jobs.iter().find(|j| j.id == job.key) {
1230 Some(spec) => (caller.clone(), spec.clone(), None),
1231 None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),
1232 },
GitHub Actions on g1t, part two: running workflows1233 };
Actions: reusable workflows in the repository1234 let spec = &spec;
GitHub Actions on g1t, part two: running workflows1235 let repo = repo_path(&run.repo);
1236 let trusted = run.trusted != 0;
Secrets and variables: one list, rows per environment, for workflows and deployments1237 // The job's `environment:`, by name: entries with a value for it give
1238 // that value instead of their default, as GitHub's environment
1239 // secrets do.
1240 let environment: Option<String> = match spec.raw.get("environment") {
1241 Some(Value::String(name)) if !name.contains("${{") => Some(name.clone()),
1242 Some(Value::Object(env)) => env.get("name").and_then(Value::as_str).filter(|n| !n.contains("${{")).map(str::to_owned),
1243 _ => None,
1244 };
1245 // G1T_TOKEN, and GITHUB_TOKEN as its alias: the workspace's own
1246 // token, for as long as the job may run.
GitHub Actions on g1t, part two: running workflows1247 let token = if trusted {
1248 match self.workspace_actor(&repo.namespace).await? {
1249 Some(workspace) => {
1250 let created: CreatedAccessToken = g1t_kit::call(
1251 &self.identity,
1252 "create_access_token",
1253 &CreateAccessTokenArgs {
1254 user: workspace,
Secrets and variables: one list, rows per environment, for workflows and deployments1255 name: format!("G1T_TOKEN for {} run {}", run.repo, run.number),
GitHub Actions on g1t, part two: running workflows1256 ttl_seconds: Some(u64::from(job.timeout_minutes) * 60 + 600),
1257 },
1258 )
1259 .await?;
1260 created.token
1261 }
1262 None => String::new(),
1263 }
1264 } else {
1265 String::new()
1266 };
Secrets and variables: one list, rows per environment, for workflows and deployments1267 // A run that is not trusted (a pull request from outside the
1268 // workspace) gets no secrets and an empty token.
1269 let mut secrets = if trusted {
1270 self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?
1271 } else {
1272 Map::new()
1273 };
1274 secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));
GitHub Actions on g1t, part two: running workflows1275 secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));
1276 let masks: Vec<String> = secrets.values().filter_map(|v| v.as_str()).filter(|v| v.len() >= 4).map(str::to_owned).collect();
Secrets and variables: one list, rows per environment, for workflows and deployments1277 let vars = self.variables_for(&run.repo_id, &run.repo, environment.as_deref(), trusted).await?;
GitHub Actions on g1t, part two: running workflows1278
1279 let jobs = self.job_rows(&run.id).await?;
1280 let mut needs = Map::new();
Actions: reusable workflows in the repository1281 // In a called workflow, its jobs' keys sit under the job that called it.
1282 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 workflows1283 for need in &spec.needs {
Actions: reusable workflows in the repository1284 let key = match &parent {
1285 Some(parent) => format!("{parent}/{need}"),
1286 None => need.clone(),
1287 };
1288 let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();
GitHub Actions on g1t, part two: running workflows1289 let mut outputs = Map::new();
1290 for row in &rows {
1291 if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {
1292 outputs.extend(more);
1293 }
1294 }
1295 needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));
1296 }
1297 let siblings = jobs.iter().filter(|row| row.key == job.key).count();
1298 let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));
1299 let info = run.info();
1300 let mut github = info.context(&job.key, &token, run.action.as_deref());
1301 github["token"] = json!(token);
1302
1303 // Where to check out: a pull request's fork, or the repository.
1304 let clone_url = match run.pull {
1305 Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" => {
1306 let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(
1307 &self.work,
1308 "get_pull",
1309 &g1t_contracts::work::ViewArgs {
1310 repo: repo.clone(),
1311 number,
1312 viewer: self.workspace_actor(&repo.namespace).await?,
1313 after_seq: 0,
1314 },
1315 )
1316 .await?;
1317 match located {
1318 Outcome::Ok(detail) => match detail.pull.fork {
1319 Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),
1320 None => format!("{SITE}/{}.git", run.repo),
1321 },
1322 Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),
1323 }
1324 }
1325 _ => format!("{SITE}/{}.git", run.repo),
1326 };
1327
1328 Ok(Outcome::Ok(json!({
1329 "job": job.id,
1330 "run": run.id,
1331 "key": job.key,
1332 "name": job.name,
1333 "spec": spec.raw,
1334 "workflow": {
1335 "env": workflow.env,
1336 "defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),
1337 },
1338 "github": github,
1339 "variables": info.variables(&job.key),
1340 "event": info.event,
1341 "contexts": {
1342 "vars": vars,
1343 "secrets": secrets,
Actions: reusable workflows in the repository1344 "inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),
GitHub Actions on g1t, part two: running workflows1345 "matrix": matrix,
1346 "needs": needs,
1347 "strategy": {
1348 "fail-fast": spec.fail_fast,
1349 "job-index": job.ordinal,
1350 "job-total": siblings,
1351 "max-parallel": spec.max_parallel.unwrap_or(siblings as u32),
1352 },
1353 "runner": runner_context(),
1354 },
1355 "checkout": {
1356 "repository": run.repo,
1357 "url": clone_url,
1358 "sha": run.sha,
1359 "ref": run.git_ref,
1360 "token": token,
1361 },
1362 "timeoutMinutes": job.timeout_minutes,
1363 "masks": masks,
1364 })))
1365 }
1366
1367 /// `job_report`: the sandbox telling how the job is going.
1368 pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {
1369 let job = check!(self.job_for_token(&a).await?);
1370 let report = &a.report;
1371 let at = now();
1372 match report["kind"].as_str().unwrap_or_default() {
1373 "steps" => {
1374 // The list can grow as the job goes (post steps), so steps
1375 // already reported keep where they stand.
1376 let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1377 let steps: Vec<Value> = report["steps"]
1378 .as_array()
1379 .map(|names| {
1380 names
1381 .iter()
1382 .enumerate()
1383 .map(|(i, name)| match known.get(i) {
1384 Some(step) if step["status"] != "queued" => step.clone(),
1385 _ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),
1386 })
1387 .collect()
1388 })
1389 .unwrap_or_default();
1390 self.db
1391 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
1392 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
1393 .run()
1394 .await?;
1395 }
1396 "step" => {
1397 let number = report["number"].as_u64().unwrap_or(0) as usize;
1398 let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();
1399 if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {
1400 let status = report["status"].as_str().unwrap_or("in_progress");
1401 step["status"] = json!(status);
1402 if status == "in_progress" {
1403 step["startedAt"] = json!(at);
1404 }
1405 if status == "completed" {
1406 step["finishedAt"] = json!(at);
1407 step["conclusion"] = report["conclusion"].clone();
1408 }
1409 if let Some(name) = report["name"].as_str() {
1410 step["name"] = json!(name);
1411 }
1412 }
1413 self.db
1414 .prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")
1415 .bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?
1416 .run()
1417 .await?;
1418 }
1419 "log" => {
1420 let mut text = report["text"].as_str().unwrap_or_default().to_owned();
1421 if text.len() > MAX_CHUNK_BYTES {
1422 let mut cut = MAX_CHUNK_BYTES;
1423 while !text.is_char_boundary(cut) {
1424 cut -= 1;
1425 }
1426 text.truncate(cut);
1427 }
1428 #[derive(Deserialize)]
1429 struct Size {
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs1430 n: Option<f64>,
1431 seq: Option<f64>,
GitHub Actions on g1t, part two: running workflows1432 }
1433 let size = self
1434 .db
1435 .prepare("SELECT SUM(LENGTH(text)) AS n, MAX(seq) AS seq FROM logs WHERE job_id = ?")
1436 .bind(&[job.id.as_str().into()])?
1437 .first::<Size>(None)
1438 .await?;
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs1439 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 workflows1440 if used < MAX_LOG_BYTES {
1441 if used + text.len() >= MAX_LOG_BYTES {
1442 text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");
1443 }
1444 self.db
1445 .prepare("INSERT INTO logs (job_id, seq, step, text) VALUES (?, ?, ?, ?)")
GitHub Actions on g1t, part three: .g1t/workflows, the pages, the docs1446 .bind(&[job.id.as_str().into(), // Numbers go to D1 as f64: a u64 would be a BigInt, which it refuses.
1447 (seq + 1.0).into(), (report["step"].as_u64().unwrap_or(0) as u32).into(), text.into()])?
GitHub Actions on g1t, part two: running workflows1448 .run()
1449 .await?;
1450 }
1451 self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;
1452 }
1453 "annotation" => {
1454 let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();
1455 if annotations.len() < MAX_ANNOTATIONS {
1456 annotations.push(json!({
1457 "level": report["level"].as_str().unwrap_or("notice"),
1458 "message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),
1459 "title": report["title"],
1460 "file": report["file"],
1461 "line": report["line"],
1462 }));
1463 self.db
1464 .prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")
1465 .bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?
1466 .run()
1467 .await?;
1468 }
1469 }
1470 "done" => {
1471 let conclusion = report["conclusion"]
1472 .as_str()
1473 .filter(|c| matches!(*c, "success" | "failure" | "cancelled"))
1474 .unwrap_or("failure");
1475 let outputs = report["outputs"].as_object().cloned();
1476 Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;
1477 }
1478 other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),
1479 }
1480 Ok(Outcome::Ok(json!({ "ok": true })))
1481 }
1482
1483 // --- Every minute ---------------------------------------------------------------
1484
1485 pub async fn on_minute(&self, now_ms: u64) -> Result<()> {
1486 let minute = now_ms / 60_000 * 60_000;
1487 if let Err(error) = self.run_schedules(minute).await {
1488 worker::console_error!("actions: schedules failed: {error}");
1489 }
1490 // Jobs whose sandbox went quiet or ran past their time.
1491 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
1492 for job in running {
1493 // Times in g1t's format compare as text.
1494 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
1495 let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());
1496 let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;
1497 let over = job.started_at.as_deref().is_some_and(|started| started < before(limit).as_str());
1498 if over {
1499 let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);
1500 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
1501 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
1502 } else if silent {
1503 self.finish_job(&job.id, "failure", Some("The runner stopped answering."), None).await?;
1504 }
1505 }
1506 self.start_queued().await
1507 }
1508}
1509