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