Sign in

flagon-io/g1t

public

g1t/services/actions/src/plan.rs

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