services/actions/src/plan.rs · flagon-io/g1t

"`{file}` would be the {}th level of workflows: a run connects at most ten, the workflow it started from and nine levels of reusable workflows below it.",

318

depth + 1

319

));

320

}

321

let mut unique: Vec<&str> = called.to_vec();

322

unique.push(file);

323

unique.sort_unstable();

324

unique.dedup();

325

if unique.len() > MAX_CALLED_WORKFLOWS {

326

return Err(format!("Calling `{file}` would make {} reusable workflows in this run: a run calls at most {MAX_CALLED_WORKFLOWS} different ones.", unique.len()));

}

Ok(())

}

/// What a job is shown as: its `name` (else its id), under the job whose

332

/// call made it (`caller`), read with its matrix when it has an

333

/// expression, else with the matrix's values after it.

334

fn shown_name(job: &workflow::Job, caller: Option<&str>, combination: &Map<String, Value>, scope: &Scope) -> String {

335

let base = job.name.clone().unwrap_or(job.id.clone());

336

let base = match caller {

337

Some(caller) => format!("{caller} / {base}"),

338

None => base,

339

};

340

if expr::has_expression(&base) {

341

expr::interpolate(&base, scope).unwrap_or(base)

342

} else if job.matrix.is_some() {

343

matrix::job_name(&base, combination)

} else {

base

}

}

/// Debug logging for a job, as a re-run with it turned on gives GitHub's:

350

/// `RUNNER_DEBUG=1` and `runner.debug`, and `ACTIONS_STEP_DEBUG` and

351

/// `ACTIONS_RUNNER_DEBUG` set, which the runner reads as the secrets of

352

/// those names (`::debug::` lines shown, and how each step's `if` read).

353

pub fn debug_logging(variables: &mut Map<String, Value>, runner: &mut Value) {

354

variables.insert("RUNNER_DEBUG".into(), json!("1"));

355

variables.insert("ACTIONS_STEP_DEBUG".into(), json!("true"));

356

variables.insert("ACTIONS_RUNNER_DEBUG".into(), json!("true"));

357

runner["debug"] = json!("1");

}

impl JobRow {

pub fn needs(&self) -> Vec<String> {

362

serde_json::from_str(&self.needs).unwrap_or_default()

363

}

364

365

pub fn call(&self) -> Option<Value> {

366

self.call.as_deref().and_then(|call| serde_json::from_str(call).ok())

367

}

368

369

/// For a job that calls a reusable workflow: what the keys of the jobs

370

/// its call makes start with (`call_prefix`).

371

pub fn call_prefix(&self) -> Option<String> {

372

let call = self.call().filter(|call| call["role"] == "caller")?;

373

Some(call["prefix"].as_str().unwrap_or(&self.key).to_owned())

374

}

375

376

/// A job that calls a reusable workflow and has not called it yet: it

377

/// waits for room under its matrix's `max-parallel`, or in its

378

/// concurrency group.

379

pub fn holds_call(&self) -> bool {

380

self.status == "waiting" && self.call_prefix().is_some()

381

}

382

383

/// For a job of a called workflow: that workflow, the job's own id in

384

/// it, and the job.

385

pub fn callee(&self) -> Option<(Workflow, workflow::Job, Value)> {

386

let call = self.call().filter(|call| call["role"] == "callee")?;

387

let called = workflow::parse(call["source"].as_str()?).ok()?;

388

let job = called.jobs.iter().find(|job| call["job"].as_str() == Some(job.id.as_str()))?.clone();

389

Some((called, job, call))

}

}

/// A job to decide on: its key, its definition, and what it needs, as

394

/// (name in `needs`, key of the jobs).

395

type Unit = (String, workflow::Job, Vec<(String, String)>);

396

397

/// How many levels of reusable workflows may sit below the workflow a run

398

/// started from, as on GitHub (ten levels of workflows in all).

399

const MAX_CALL_DEPTH: u64 = 9;

400

/// How many different reusable workflows a run may call, as on GitHub.

401

const MAX_CALLED_WORKFLOWS: usize = 50;

402

403

/// What the jobs of one key came to, for `needs.<key>`.

404

fn key_result(rows: &[&JobRow]) -> &'static str {

405

let failed = |row: &&&JobRow| row.conclusion.as_deref() == Some("failure") && row.continue_on_error == 0;

406

if rows.iter().any(|row| failed(&row)) {

407

"failure"

408

} else if rows.iter().any(|row| row.conclusion.as_deref() == Some("cancelled")) {

409

"cancelled"

410

} else if rows.iter().all(|row| row.conclusion.as_deref() == Some("skipped")) {

"skipped"

} else {

"success"

}

}

/// Whether any job before `key` failed: one it needs, or one those need,

418

/// however far back. A job after a skipped one still sees the failure

419

/// that skipped it, as GitHub's failure() does. `needs_of`: each key's

420

/// needs, as keys; `failed`: whether a key's jobs came to a failure.

421

fn ancestor_failed(needs_of: &std::collections::HashMap<&str, Vec<&str>>, key: &str, failed: impl Fn(&str) -> bool) -> bool {

422

let mut seen = std::collections::HashSet::new();

423

let mut stack: Vec<&str> = needs_of.get(key).cloned().unwrap_or_default();

424

while let Some(next) = stack.pop() {

425

if !seen.insert(next) {

continue;

}

if failed(next) {

return true;

}

stack.extend(needs_of.get(next).into_iter().flatten().copied());

}

false

}

/// A job's `environment:`, as deployments read it: its name, the address in

437

/// `environment.url` once its expressions are filled in from the run, and

438

/// whether the job deploys to it. A job with `deployment: false` only reads

439

/// the environment's secrets and variables, and makes no deployment.

440

#[derive(Clone, Debug, Default, PartialEq)]

441

pub(crate) struct JobEnvironment {

442

pub(crate) name: String,

443

pub(crate) url: Option<String>,

444

pub(crate) deploys: bool,

445

}

446

447

/// The environment `environment:` names, as written in `raw` (a job), with

448

/// `contexts` to fill in expressions: null when it names none, or only

449

/// through an expression this cannot read before the job runs.

450

pub(crate) fn environment_of(raw: &Value, contexts: &Map<String, Value>) -> Option<JobEnvironment> {

451

let scope = Scope { contexts, status: Status::Success, hash_files: None };

452

let plain = |value: &Value| -> Option<String> {

453

let text = match value {

454

Value::String(text) if text.contains("${{") => expr::interpolate_value(value, &scope).ok().map(|v| expr::to_text(&v))?,

455

Value::String(text) => text.clone(),

456

_ => return None,

457

};

458

let text = text.trim().to_owned();

459

(!text.is_empty()).then_some(text)

460

};

461

match raw.get("environment")? {

462

name @ Value::String(_) => Some(JobEnvironment { name: plain(name)?, url: None, deploys: workflow::records_deployment(raw) }),

463

Value::Object(env) => {

464

let name = plain(env.get("name")?)?;

let url = env

.get("url")

.and_then(plain)

.filter(|url| url.starts_with("https://") || url.starts_with("http://"));

469

Some(JobEnvironment { name, url, deploys: workflow::records_deployment(raw) })

}

_ => None,

}

}

/// The contexts a job's `runs-on` and `environment` are read with before it

476

/// runs: the run's `github`, its inputs, and the job's matrix.

477

fn start_contexts(run: &RunRow, job: &JobRow, key: &str) -> Map<String, Value> {

478

let mut contexts = Map::new();

479

contexts.insert("github".into(), run.info().context(key, "", run.action.as_deref()));

480

contexts.insert("inputs".into(), Value::Object(run.inputs()));

481

contexts.insert("matrix".into(), job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or_else(|| json!({})));

contexts

}

/// The job as its workflow (or the workflow it calls) defines it.

486

fn job_spec(run: &RunRow, job: &JobRow) -> Option<workflow::Job> {

487

match job.callee() {

488

Some((_, spec, _)) => Some(spec),

489

None => workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key)),

}

}

/// The environment a job deploys to, if it deploys: by the name read when

494

/// its needs were done (which may have needed their outputs), else as

495

/// its `environment:` reads now.

496

pub(crate) fn deploys_to(run: &RunRow, job: &JobRow) -> Option<JobEnvironment> {

497

let spec = job_spec(run, job)?;

498

let read = environment_of(&spec.raw, &start_contexts(run, job, &spec.id));

499

let env = match (read, &job.environment) {

500

(Some(env), Some(name)) => Some(JobEnvironment { name: name.clone(), ..env }),

501

(Some(env), None) => Some(env),

502

(None, Some(name)) => Some(JobEnvironment { name: name.clone(), url: None, deploys: workflow::records_deployment(&spec.raw) }),

503

(None, None) => None,

504

};

505

env.filter(|env| env.deploys)

506

}

507

508

/// Whether a run's repository was public when the run was made, from its

509

/// event's `repository.private`: private when the event doesn't say.

510

fn run_is_public(run: &RunRow) -> bool {

511

run.info().event.get("repository").and_then(|repository| repository.get("private")).and_then(Value::as_bool) == Some(false)

512

}

513

514

/// The machine a job's `runs-on` asks for (`instance_for`), for the

515

/// runner: none for the standard one. `ubuntu-latest` follows GitHub: the

516

/// 4-core machine in a public repository. Billing charges the job as the

517

/// machine named here, so what it ran on is what it pays for.

518

fn start_instance(run: &RunRow, job: &JobRow) -> Option<String> {

519

let spec = match job.callee() {

520

Some((_, spec, _)) => spec,

521

None => workflow::parse(&run.source).ok().and_then(|w| w.jobs.into_iter().find(|j| j.id == job.key))?,

522

};

523

// `runs-on` as the job was queued with: its matrix and the run's

524

// inputs. A label that needs more than those is the standard machine.

525

let contexts = start_contexts(run, job, &spec.id);

526

let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };

527

let labels: Vec<String> = match expr::interpolate_value(&spec.runs_on, &scope).unwrap_or(Value::Null) {

528

Value::String(label) => vec![label],

529

Value::Array(labels) => labels.iter().map(expr::to_text).collect(),

530

Value::Object(given) => match given.get("labels") {

531

Some(Value::Array(labels)) => labels.iter().map(expr::to_text).collect(),

532

Some(label) => vec![expr::to_text(label)],

None => Vec::new(),

},

_ => Vec::new(),

};

let instance = g1t_contracts::actions::instance_for(&labels, run_is_public(run));

538

(instance != g1t_contracts::actions::STANDARD_INSTANCE).then(|| instance.label.to_owned())

}

fn now() -> String {

rfc3339(now_ms())

}

/// How a finished run went for one environment, from the conclusions of

546

/// the jobs that deploy to it: failed if any failed, an error if any was

547

/// cancelled, else a success; nothing when none ran (all skipped).

548

pub(crate) fn deployment_outcome(conclusions: &[Option<String>]) -> Option<&'static str> {

549

let ran: Vec<&str> = conclusions.iter().flatten().map(String::as_str).filter(|c| *c != "skipped").collect();

if ran.is_empty() {

return None;

}

if ran.iter().any(|c| *c == "failure" || *c == "timed_out") {

554

return Some("failure");

555

}

556

if ran.contains(&"cancelled") {

557

return Some("error");

}

Some("success")

}

impl Actions {

pub async fn run_row(&self, id: &str) -> Result<Option<RunRow>> {

564

self.db.prepare("SELECT * FROM runs WHERE id = ?").bind(&[id.into()])?.first::<RunRow>(None).await

565

}

566

567

pub async fn job_rows(&self, run_id: &str) -> Result<Vec<JobRow>> {

568

self.db

569

.prepare("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid")

570

.bind(&[run_id.into()])?

.all()

.await?

.results::<JobRow>()

}

pub async fn run_summary(&self, id: &str) -> Result<Outcome<WorkflowRun>> {

577

Ok(match self.run_row(id).await? {

578

Some(row) => Outcome::Ok(row.summary()),

579

None => fail(FailureCode::NotFound, "No such run."),

})

}

/// The contexts every expression outside a job's steps may use.

584

fn base_contexts(run: &RunRow, vars: &Map<String, Value>, job: &str) -> Map<String, Value> {

585

let mut contexts = Map::new();

586

contexts.insert("github".into(), run.info().context(job, "", run.action.as_deref()));

587

contexts.insert("inputs".into(), Value::Object(run.inputs()));

588

contexts.insert("vars".into(), Value::Object(vars.clone()));

589

contexts.insert("needs".into(), json!({}));

590

contexts.insert("runner".into(), runner_context());

contexts

}

/// Whether `namespace` is a person's own workspace: usernames and

595

/// workspaces share one namespace, so it is when someone has that

596

/// username. Kept in the isolate for a few minutes, so a burst of

597

/// runs asks identity once.

598

async fn personal_namespace(&self, namespace: &str) -> Result<bool> {

599

const KEEP_MS: u64 = 5 * 60 * 1000;

600

thread_local! {

601

static PERSONAL: std::cell::RefCell<std::collections::HashMap<String, (bool, u64)>> = Default::default();

602

}

603

let now = now_ms();

604

let key = namespace.to_ascii_lowercase();

605

if let Some(personal) = PERSONAL.with(|kept| kept.borrow().get(&key).filter(|(_, at)| now.saturating_sub(*at) < KEEP_MS).map(|(personal, _)| *personal)) {

606

return Ok(personal);

607

}

608

let found: g1t_contracts::Viewer = g1t_kit::call(

609

&self.identity,

610

"user_by_username",

611

&g1t_contracts::identity::UsernameArgs { username: key.clone() },

612

)

613

.await?;

614

let personal = found.is_some_and(|user| user.username.eq_ignore_ascii_case(namespace));

615

PERSONAL.with(|kept| {

616

let mut kept = kept.borrow_mut();

617

if kept.len() > 1000 {

618

kept.clear();

619

}

620

kept.insert(key, (personal, now));

});

Ok(personal)

}

/// Makes a run and its jobs, and starts what can start. `None` when the

626

/// event already started this workflow.

627

pub async fn create_run(&self, new: NewRun) -> Result<Option<String>> {

628

self.create_run_with(new, false).await

629

}

630

631

/// As `create_run`, but the run's pending status must reach Work: one

632

/// that cannot is cancelled and the error returned, so that a caller

633

/// waiting on statuses (the merge queue) never reads "no workflows" or

634

/// a premature pass from a run nothing was told about.

635

pub async fn create_run_registered(&self, new: NewRun) -> Result<Option<String>> {

636

self.create_run_with(new, true).await

637

}

638

639

async fn create_run_with(&self, new: NewRun, strict: bool) -> Result<Option<String>> {

640

// Nothing starts on an archived repository. A deleted one is never

641

// found to start anything on.

642

if new.repo.archived() {

643

return Ok(None);

644

}

645

// Nor while its Actions are off (Settings, Actions; a fork's are

646

// until someone turns them on), whatever started it. This is the

647

// gate; read once per request, so the runs of one push share it.

648

if !self.actions_on(&new.repo).await? {

649

return Ok(None);

650

}

651

let workflow_row = self.workflow_row(&new.repo, &new.path, &new.workflow.display_name(&new.path), &new.source).await?;

652

let numbered = self

653

.db

654

.prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")

655

.bind(&[workflow_row.id.as_str().into()])?

656

.first::<Count>(None)

657

.await?

658

.map_or(1, |count| count.n);

659

let id = new_id("run", now_ms());

660

let mut info = new.info.clone();

661

// A personal workspace's repositories are a `User`'s, as on GitHub.

662

if self.personal_namespace(&new.repo.namespace).await? {

663

crate::payload::owned_by_person(&mut info.event, &new.repo.namespace);

664

}

665

info.workflow = new.workflow.display_name(&new.path);

666

info.workflow_path = new.path.clone();

667

info.run_id = id.clone();

668

info.run_number = u64::from(numbered);

669

let vars = self

670

.variables_for(&new.repo.id, &format!("{}/{}", new.repo.namespace, new.repo.name), None, new.trusted)

671

.await?;

672

673

// run-name and the concurrency group read github, inputs and vars.

674

let mut contexts = Map::new();

675

contexts.insert("github".into(), info.context("", "", new.action.as_deref()));

676

contexts.insert("inputs".into(), Value::Object(new.inputs.clone()));

677

contexts.insert("vars".into(), Value::Object(vars));

678

let scope = Scope {

679

contexts: &contexts,

680

status: Status::Success,

hash_files: None,

};

let title = new

.workflow

.run_name

.as_deref()

.and_then(|run_name| expr::interpolate(run_name, &scope).ok())

688

.filter(|title| !title.trim().is_empty())

689

.unwrap_or(new.title.clone());

690

let group = new.workflow.concurrency.as_ref().and_then(|c| expr::interpolate(&c.group, &scope).ok());

691

let cancel_in_progress = new

.workflow

.concurrency

.as_ref()

.and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())

696

.is_some_and(|value| expr::truthy(&value));

697

698

let approval = new.approval.as_ref().map(|reason| RunApproval { state: "required".into(), reason: reason.clone(), approved_by: None });

let inserted = self

.db

.prepare(

"INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, action, git_ref, sha,

703

pull, status, actor, actor_id, source, info, inputs, trusted, concurrency_group, event_key, created_at,

704

approval, cancel_in_progress)

705

VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",

)

.bind(&[

id.as_str().into(),

workflow_row.id.as_str().into(),

710

new.repo.id.as_str().into(),

711

format!("{}/{}", new.repo.namespace, new.repo.name).into(),

712

new.path.as_str().into(),

713

info.workflow.as_str().into(),

714

title.as_str().into(),

715

numbered.into(),

716

info.event_name.as_str().into(),

717

optional(new.action.as_deref()),

718

info.git_ref.as_str().into(),

719

new.head_sha.as_deref().unwrap_or(&info.sha).into(),

720

new.pull.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),

721

if approval.is_some() { "action_required" } else { "queued" }.into(),

722

optional(new.actor.as_deref()),

723

optional(new.actor_id.as_deref()),

724

new.source.as_str().into(),

725

serde_json::to_string(&info)?.into(),

726

serde_json::to_string(&new.inputs)?.into(),

727

u32::from(new.trusted).into(),

728

optional(group.as_deref()),

729

new.event_key.as_str().into(),

730

now().into(),

731

approval.as_ref().map(serde_json::to_string).transpose()?.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),

732

u32::from(cancel_in_progress).into(),

733

])?

734

.first::<Value>(None)

735

.await?;

736

if inserted.is_none() {

737

return Ok(None);

738

}

739

let mut unregistered = None;

740

if let Some(run) = self.run_row(&id).await? {

741

if strict {

742

if let Err(error) = self.try_report_pending(&run).await {

743

unregistered = Some((run, error));

744

}

745

} else {

746

self.report_pending(&run).await?;

}

}

// Every job, waiting; each is expanded when the jobs it needs are done.

751

let mut statements = Vec::new();

752

for job in &new.workflow.jobs {

statements.push(

self.db

.prepare(

"INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting')",

757

)

758

.bind(&[

759

new_id("job", now_ms()).into(),

760

id.as_str().into(),

761

new.repo.id.as_str().into(),

762

new.repo.namespace.as_str().into(),

763

job.id.as_str().into(),

764

job.name.clone().filter(|n| !expr::has_expression(n)).unwrap_or(job.id.clone()).into(),

765

serde_json::to_string(&job.needs)?.into(),

])?,

);

}

self.db.batch(statements).await?;

770

if let Some((run, error)) = unregistered {

771

self.cancel_run(&run, "Its pending status could not be registered.", false).await?;

return Err(error);

}

// A pull request's run from outside waits to be approved; it joins

776

// its concurrency group once it is (protection.rs, approve_run).

777

if approval.is_some() {

778

return Ok(Some(id));

779

}

780

if let Some(run) = self.run_row(&id).await? {

781

self.enter_group(&run).await?;

}

Ok(Some(id))

}

/// A run about to start: one run at a time per concurrency group, as on

787

/// GitHub. A newer run replaces one that waits in the group, and with

788

/// `cancel-in-progress` the one that runs; otherwise it waits as

789

/// `pending` for the one that runs. Then it moves along.

790

pub(crate) async fn enter_group(&self, run: &RunRow) -> Result<()> {

791

if let Some(group) = &run.concurrency_group {

792

let cancel_in_progress = run.cancel_in_progress != 0;

let others = self

.db

.prepare(

"SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND id != ? AND status NOT IN ('completed', 'action_required') ORDER BY id",

797

)

798

.bind(&[run.repo_id.as_str().into(), group.as_str().into(), run.id.as_str().into()])?

799

.all()

800

.await?

801

.results::<RunRow>()?;

802

for other in &others {

803

if cancel_in_progress || other.status == "pending" {

804

// A newer run replaces a waiting one, as on GitHub.

805

self.cancel_run(other, "A newer run in the same concurrency group replaced it.", false).await?;

806

}

807

}

808

if !cancel_in_progress && others.iter().any(|other| other.status != "pending") {

809

self.db

810

.prepare("UPDATE runs SET status = 'pending' WHERE id = ?")

811

.bind(&[run.id.as_str().into()])?

.run()

.await?;

return Ok(());

}

}

self.advance(&run.id).await

818

}

819

820

/// A run that could not start, such as for a workflow file that does

821

/// not read. None while `repo`'s Actions are off, as for `create_run`.

822

#[allow(clippy::too_many_arguments)]

823

pub async fn record_failed_run(

&self,

repo: &Repo,

row: &WorkflowRow,

git_ref: &str,

sha: &str,

event_key: &str,

actor_id: Option<&str>,

actor: &str,

problem: &str,

) -> Result<()> {

if !self.actions_on(repo).await? {

return Ok(());

}

let numbered = self

.db

.prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")

840

.bind(&[row.id.as_str().into()])?

841

.first::<Count>(None)

842

.await?

843

.map_or(1, |count| count.n);

let at = now();

self.db

.prepare(

"INSERT OR IGNORE INTO runs (id, workflow_id, repo_id, repo, path, name, title, number, event, git_ref, sha, status,

848

conclusion, error, actor, actor_id, source, info, event_key, created_at, finished_at)

849

VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'push', ?, ?, 'completed', 'failure', ?, ?, ?, ?, '{}', ?, ?, ?)",

850

)

851

.bind(&[

852

new_id("run", now_ms()).into(),

853

row.id.as_str().into(),

854

row.repo_id.as_str().into(),

855

row.repo.as_str().into(),

856

row.path.as_str().into(),

857

row.path.as_str().into(),

858

"Invalid workflow file".into(),

numbered.into(),

git_ref.into(),

sha.into(),

problem.into(),

actor.into(),

optional(actor_id),

row.source.as_str().into(),

event_key.into(),

at.as_str().into(),

at.as_str().into(),

])?

.run()

.await?;

Ok(())

}

/// Moves a run along: jobs whose needs are done are decided on, and

876

/// jobs that can start are started.

877

pub async fn advance(&self, run_id: &str) -> Result<()> {

878

// Each pass may finish jobs (skipped ones), which may free others.

879

for _ in 0..20 {

880

let Some(run) = self.run_row(run_id).await? else { return Ok(()) };

881

if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {

882

return Ok(());

883

}

884

let workflow = match workflow::parse(&run.source) {

885

Ok(workflow) => workflow,

886

Err(problem) => {

887

self.finish_run(&run, Some(&problem)).await?;

return Ok(());

}

};

let jobs = self.job_rows(run_id).await?;

892

let mut changed = false;

893

// What to decide on: the workflow's jobs, and the jobs of the

894

// workflows they call, each with its needs as (name, key).

895

let mut units: Vec<Unit> = workflow

896

.jobs

897

.iter()

898

.map(|job| (job.id.clone(), job.clone(), job.needs.iter().map(|n| (n.clone(), n.clone())).collect()))

899

.collect();

900

let mut seen = std::collections::HashSet::new();

901

for row in &jobs {

902

if !seen.insert(row.key.clone()) {

903

continue;

904

}

905

if let Some((_, job, call)) = row.callee() {

906

let parent = call["parent"].as_str().unwrap_or_default().to_owned();

907

let needs = job.needs.iter().map(|n| (n.clone(), format!("{parent}/{n}"))).collect();

908

units.push((row.key.clone(), job, needs));

909

}

910

}

911

// Each key's needs, as keys, to look back through for failure().

912

let needs_of: std::collections::HashMap<&str, Vec<&str>> =

913

units.iter().map(|(key, _, needs)| (key.as_str(), needs.iter().map(|(_, need)| need.as_str()).collect())).collect();

914

let key_failed = |key: &str| key_result(&jobs.iter().filter(|row| row.key == key).collect::<Vec<_>>()) == "failure";

915

for (key, job, needs) in &units {

916

let rows: Vec<&JobRow> = jobs.iter().filter(|row| &row.key == key).collect();

917

if rows.is_empty() || !rows.iter().all(|row| row.status == "waiting" && !row.holds_call()) {

918

continue;

919

}

920

let needed: Vec<(&String, Vec<&JobRow>)> =

921

needs.iter().map(|(name, need)| (name, jobs.iter().filter(|row| &row.key == need).collect())).collect();

922

if !needed.iter().all(|(_, rows)| rows.iter().all(|row| row.status == "completed")) {

923

continue;

924

}

925

let failed_before = ancestor_failed(&needs_of, key, key_failed);

926

self.decide(&run, job, rows[0], &needed, failed_before).await?;

927

changed = true;

928

}

929

// A job that called a workflow finishes with that workflow's jobs.

930

for row in jobs.iter().filter(|row| row.status == "calling") {

931

let prefix = row.call_prefix().unwrap_or(row.key.clone());

932

let children: Vec<&JobRow> = jobs

933

.iter()

934

.filter(|child| child.call().is_some_and(|call| call["role"] == "callee" && call["parent"].as_str() == Some(prefix.as_str())))

935

.collect();

936

if !children.is_empty() && children.iter().all(|child| child.status == "completed") {

937

self.finish_call(&run, row, &children).await?;

changed = true;

}

}

// Calls that wait start while their matrix's `max-parallel` and

942

// their concurrency group have room.

943

let mut calling: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();

944

for row in jobs.iter().filter(|row| row.status == "calling") {

945

*calling.entry(row.key.as_str()).or_default() += 1;

946

}

947

for row in jobs.iter().filter(|row| row.holds_call()) {

948

let running = calling.get(row.key.as_str()).copied().unwrap_or(0);

949

if row.max_parallel.is_some_and(|max| running >= max) || self.group_busy(row).await? {

950

continue;

951

}

952

if self.start_call(row).await? {

953

calling.insert(row.key.as_str(), running + 1);

changed = true;

}

}

if !changed {

break;

}

}

self.start_queued().await?;

962

self.finish_if_done(run_id).await

963

}

964

965

/// Decides on one job whose needs are done: skip it, fail it, or expand

966

/// it into its matrix and queue it. `failed_before`: whether any job

967

/// before it failed, however far back (`ancestor_failed`).

968

async fn decide(&self, run: &RunRow, job: &workflow::Job, row: &JobRow, needed: &[(&String, Vec<&JobRow>)], failed_before: bool) -> Result<()> {

969

let vars = self.variables_for(&run.repo_id, &run.repo, None, run.trusted != 0).await?;

970

let mut contexts = Self::base_contexts(run, &vars, &job.id);

971

// A called workflow's jobs read the inputs they were called with.

972

let call = row.call();

973

// Shown under the job whose call made it, by that job's name.

974

let caller = call.as_ref().filter(|c| c["role"] == "callee").and_then(|c| {

975

c["title"].as_str().map(str::to_owned).or_else(|| c["parent"].as_str().map(|parent| parent.replace('/', " / ")))

976

});

977

if let Some(call) = call.as_ref().filter(|c| c["role"] == "callee") {

978

contexts.insert("inputs".into(), call["inputs"].clone());

979

}

980

let mut needs = Map::new();

981

let mut results = Vec::new();

982

for (key, rows) in needed {

983

let result = key_result(rows);

984

let mut outputs = Map::new();

985

for row in rows {

986

if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {

987

outputs.extend(more);

988

}

989

}

990

results.push(result);

991

needs.insert((*key).clone(), json!({ "result": result, "outputs": outputs }));

992

}

993

// As on GitHub: a need that was skipped makes success() false but

994

// not failure(); failure() is a failure anywhere before the job.

995

let status = expr::job_status(results, failed_before, run.conclusion.as_deref() == Some("cancelled"));

996

contexts.insert("needs".into(), Value::Object(needs));

let scope = Scope {

contexts: &contexts,

status,

hash_files: None,

};

let condition = job.condition.as_deref().unwrap_or_default();

1003

match expr::condition(condition, &scope) {

1004

Ok(true) => {}

1005

Ok(false) => return self.skip_job(row, None).await,

1006

Err(problem) => return self.fail_job(row, &format!("Its `if` does not read: {problem}")).await,

1007

}

1008

// Its matrix, which may come from a needed job's outputs.

1009

let combinations = match &job.matrix {

1010

None => vec![Map::new()],

1011

Some(matrix) => {

1012

let value = match expr::interpolate_value(matrix, &scope) {

1013

Ok(value) => value,

1014

Err(problem) => return self.fail_job(row, &format!("Its matrix does not read: {problem}")).await,

1015

};

1016

match matrix::expand(&value) {

1017

Ok(combinations) if !combinations.is_empty() => combinations,

1018

Ok(_) => return self.fail_job(row, "Its matrix makes no jobs.").await,

1019

Err(problem) => return self.fail_job(row, &problem).await,

}

}

};

// A job that calls a reusable workflow calls it once for each.

1024

if let Some(uses) = &job.uses {

1025

return self.call_workflow(run, job, row, uses, &contexts, &combinations, caller.as_deref()).await;

1026

}

1027

let total = combinations.len();

1028

let raw = job.raw.as_object().cloned().unwrap_or_default();

1029

// Whether pull requests from forks may use self-hosted runners here,

1030

// asked once, and only for a run that is not trusted.

1031

let mut forks_allowed: Option<bool> = None;

1032

// Each concurrency group its jobs join, and whether it cancels.

1033

let mut groups: Vec<(String, bool)> = Vec::new();

1034

let mut statements = Vec::new();

1035

for (index, combination) in combinations.iter().enumerate() {

1036

let mut contexts = contexts.clone();

1037

contexts.insert("matrix".into(), Value::Object(combination.clone()));

1038

contexts.insert(

1039

"strategy".into(),

1040

json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),

);

let scope = Scope {

contexts: &contexts,

status: Status::Success,

1045

hash_files: None,

1046

};

1047

let name = shown_name(job, caller.as_deref(), combination, &scope);

1048

let runs_on = expr::interpolate_value(&job.runs_on, &scope).unwrap_or(Value::Null);

1049

let labels = match &runs_on {

1050

Value::String(label) => vec![label.clone()],

1051

Value::Array(labels) => labels.iter().map(expr::to_text).collect(),

1052

Value::Object(spec) => spec.get("labels").map(|l| match l {

1053

Value::Array(labels) => labels.iter().map(expr::to_text).collect(),

1054

other => vec![expr::to_text(other)],

1055

}).unwrap_or_default(),

1056

_ => Vec::new(),

1057

};

1058

// `runs-on: self-hosted` (or a group): the workspace's own

1059

// machines, which may run any OS. Otherwise g1t's Linux sandboxes.

1060

let group = match &runs_on {

1061

Value::Object(spec) => spec.get("group").map(expr::to_text),

1062

_ => None,

1063

};

1064

let wanted = Wanted::of(&labels, group.as_deref());

1065

let mut reason = if wanted.self_hosted {

None

} else {

labels

.iter()

.find(|label| {

let lower = label.to_ascii_lowercase();

1072

lower.contains("windows") || lower.contains("macos")

1073

})

1074

.map(|label| {

1075

let os = if label.to_ascii_lowercase().contains("windows") { "windows" } else { "macos" };

1076

format!("`runs-on: {label}`: g1t's own runners are Linux. A self-hosted runner can run it: `runs-on: [self-hosted, {os}]`.")

1077

})

1078

};

1079

// A pull request from a fork runs code anyone could write: never

1080

// on the workspace's machines unless it said they may.

1081

if wanted.self_hosted && run.trusted == 0 {

1082

let allowed = match forks_allowed {

1083

Some(allowed) => allowed,

1084

None => {

1085

let allowed = self.effective_runner_settings(&row.namespace, Some(&run.repo_id)).await?.fork_pull_requests;

1086

forks_allowed = Some(allowed);

allowed

}

};

if !allowed {

reason = Some(

"Pull requests from forks do not run on self-hosted runners here. An admin can allow it under Settings, Runners.".to_owned(),

);

}

}

let max_minutes = if wanted.self_hosted { SELF_HOSTED_MAX_TIMEOUT_MINUTES } else { MAX_TIMEOUT_MINUTES };

1097

let timeout = raw

1098

.get("timeout-minutes")

1099

.and_then(|value| expr::interpolate_value(value, &scope).ok())

1100

.and_then(|value| value.as_f64().or_else(|| expr::to_text(&value).parse().ok()))

1101

.map_or(g1t_actions::workflow::DEFAULT_TIMEOUT_MINUTES.min(max_minutes), |minutes| (minutes.ceil() as u32).clamp(1, max_minutes));

1102

let continue_on_error = raw

1103

.get("continue-on-error")

1104

.and_then(|value| expr::interpolate_value(value, &scope).ok())

1105

.is_some_and(|value| expr::truthy(&value));

1106

let (mut status, mut conclusion, mut finished) = match &reason {

1107

Some(_) => ("completed", Some("failure"), Some(now())),

1108

None => ("queued", None, None),

1109

};

1110

// A self-hosted job waits, saying for what, until a runner takes it.

1111

let (labels_json, queued_at) = if wanted.self_hosted && reason.is_none() {

1112

reason = Some(waiting_reason(&wanted));

1113

(Some(serde_json::to_string(&wanted.stored())?), Some(now()))

} else {

(None, None)

};

// The environment it names, an expression read now, and what

1118

// that environment's protection rules say: it starts, waits as

1119

// `pending` (protection.rs), or may not deploy there.

1120

let environment = environment_of(&job.raw, &contexts).map(|env| env.name);

1121

if status == "queued"

1122

&& let Some(name) = &environment

1123

{

1124

match self.gate(run, name).await? {

Gate::Open => {}

Gate::Held(why) => {

status = "pending";

reason = Some(why);

}

Gate::Refused(why) => {

1131

(status, conclusion, finished) = ("completed", Some("failure"), Some(now()));

reason = Some(why);

}

}

}

// Its own concurrency group.

let job_group = job

.concurrency

.as_ref()

.and_then(|c| expr::interpolate(&c.group, &scope).ok())

1141

.map(|group| group.trim().to_owned())

1142

.filter(|group| !group.is_empty());

let job_cancel = job

.concurrency

.as_ref()

.and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())

1147

.is_some_and(|value| expr::truthy(&value));

1148

if status != "completed"

1149

&& let Some(group) = &job_group

1150

&& !groups.iter().any(|(known, _)| known == group)

1151

{

1152

groups.push((group.clone(), job_cancel));

1153

}

1154

let values: Vec<worker::wasm_bindgen::JsValue> = vec![

1155

name.into(),

1156

serde_json::to_string(combination)?.into(),

1157

status.into(),

1158

optional(conclusion),

1159

optional(reason.as_deref()),

1160

timeout.into(),

1161

u32::from(continue_on_error).into(),

1162

job.max_parallel.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),

1163

optional(finished.as_deref()),

1164

optional(labels_json.as_deref()),

1165

optional(queued_at.as_deref()),

1166

optional(environment.as_deref()),

1167

optional(job_group.as_deref()),

1168

u32::from(job_cancel).into(),

1169

];

1170

if index == 0 {

1171

let mut bound = values;

1172

bound.push(row.id.as_str().into());

statements.push(

self.db

.prepare(

"UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, timeout_minutes = ?,

1177

continue_on_error = ?, max_parallel = ?, finished_at = ?, labels = ?, queued_at = ?, environment = ?,

1178

concurrency_group = ?, cancel_in_progress = ? WHERE id = ?",

)

.bind(&bound)?,

);

} else {

let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![

1184

new_id("job", now_ms()).into(),

1185

row.run_id.as_str().into(),

1186

row.repo_id.as_str().into(),

1187

row.namespace.as_str().into(),

1188

row.key.as_str().into(),

1189

(index as u32).into(),

1190

row.needs.as_str().into(),

1191

optional(row.call.as_deref()),

1192

];

1193

bound.extend(values);

statements.push(

self.db

.prepare(

"INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, call, name, matrix, status, conclusion, reason,

1198

timeout_minutes, continue_on_error, max_parallel, finished_at, labels, queued_at, environment,

1199

concurrency_group, cancel_in_progress)

1200

VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",

)

.bind(&bound)?,

);

}

}

self.db.batch(statements).await?;

1207

self.replace_in_groups(run, &row.key, &groups).await

1208

}

1209

1210

/// A job queued in its own concurrency group: with `cancel-in-progress`

1211

/// it cancels the group's other jobs; otherwise it replaces any that

1212

/// are queued and not started, and waits for the one running

1213

/// (`start_queued` starts one of a group at a time). As on GitHub.

1214

async fn replace_in_groups(&self, run: &RunRow, key: &str, groups: &[(String, bool)]) -> Result<()> {

1215

if groups.is_empty() {

1216

return Ok(());

1217

}

1218

#[derive(Deserialize)]

struct Id {

id: String,

}

let mine: Vec<String> = self

1223

.db

1224

.prepare("SELECT id FROM jobs WHERE run_id = ? AND key = ?")

1225

.bind(&[run.id.as_str().into(), key.into()])?

.all()

.await?

.results::<Id>()?

.into_iter()

.map(|row| row.id)

.collect();

let mut moved: Vec<String> = Vec::new();

1233

for (group, cancel) in groups {

1234

let others = self

1235

.db

1236

// A call that waits in the group, or calls, is in it too.

1237

.prepare("SELECT * FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('waiting', 'queued', 'pending', 'in_progress', 'calling')")

1238

.bind(&[run.repo_id.as_str().into(), group.as_str().into()])?

1239

.all()

1240

.await?

1241

.results::<JobRow>()?;

1242

for other in others.iter().filter(|other| !mine.contains(&other.id)) {

1243

if *cancel || matches!(other.status.as_str(), "queued" | "waiting") {

1244

self.stop_job(other, "A newer job in the same concurrency group replaced it.").await?;

1245

if !moved.contains(&other.run_id) {

1246

moved.push(other.run_id.clone());

}

}

}

}

for other_run in moved.iter().filter(|id| **id != run.id) {

1252

Box::pin(self.advance(other_run)).await?;

}

Ok(())

}

/// A job that calls a reusable workflow, in the repository or another

1258

/// (reach.rs): once for each combination of its matrix, as on GitHub,

1259

/// each with the inputs and secrets it passes read with that

1260

/// combination's `matrix`. Each call waits (`holds_call`) until its

1261

/// matrix's `max-parallel` and its concurrency group have room, when

1262

/// `start_call` puts the workflow's jobs in the run under it.

1263

#[allow(clippy::too_many_arguments)]

1264

async fn call_workflow(

&self,

run: &RunRow,

job: &workflow::Job,

row: &JobRow,

uses: &str,

contexts: &Map<String, Value>,

1271

combinations: &[Map<String, Value>],

1272

caller: Option<&str>,

1273

) -> Result<()> {

1274

let depth = row.call().and_then(|c| c["depth"].as_u64()).unwrap_or(0) + 1;

1275

if depth > MAX_CALL_DEPTH {

1276

return self.fail_job(row, &call_allowed(depth, &[], uses).unwrap_err()).await;

1277

}

1278

let (file, source, origin) = match self.called_workflow(run, row, uses).await? {

1279

Ok(found) => found,

1280

Err(why) => return self.fail_job(row, &why).await,

1281

};

1282

let jobs = self.job_rows(&run.id).await?;

1283

// Each call so far, by the workflow it called.

1284

let called_paths: Vec<String> = jobs

1285

.iter()

1286

.filter(|other| other.call_prefix().is_some())

1287

.filter_map(JobRow::call)

1288

.filter_map(|call| call["path"].as_str().map(str::to_owned))

1289

.collect();

1290

let called_already: Vec<&str> = called_paths.iter().map(String::as_str).collect();

1291

if let Err(why) = call_allowed(depth, &called_already, &file) {

1292

return self.fail_job(row, &why).await;

1293

}

1294

let called = match workflow::parse(&source) {

1295

Ok(called) => called,

1296

Err(problem) => return self.fail_job(row, &format!("`{file}` does not read: {problem}")).await,

1297

};

1298

let Some(trigger) = called.trigger("workflow_call") else {

1299

return self.fail_job(row, &format!("`{file}` cannot be called: it has no `on: workflow_call`.")).await;

1300

};

1301

// Secrets: none but the job's token unless `secrets:` passes them,

1302

// by name or with `inherit`; read when each job starts (job_spec).

1303

let outer = row.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());

1304

let total = combinations.len();

1305

let mut groups: Vec<(String, bool)> = Vec::new();

1306

let mut statements = Vec::new();

1307

for (index, combination) in combinations.iter().enumerate() {

1308

let mut contexts = contexts.clone();

1309

contexts.insert("matrix".into(), Value::Object(combination.clone()));

1310

contexts.insert(

1311

"strategy".into(),

1312

json!({ "fail-fast": job.fail_fast, "job-index": index, "job-total": total, "max-parallel": job.max_parallel.unwrap_or(total as u32) }),

1313

);

1314

let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };

1315

let name = shown_name(job, caller, combination, &scope);

1316

// Inputs: what `with:` passes, as the called workflow types them.

1317

let given = match job.raw.get("with") {

1318

Some(with) => match expr::interpolate_value(with, &scope) {

1319

Ok(Value::Object(given)) => Ok(given),

1320

Ok(_) => Ok(Map::new()),

1321

Err(problem) => Err(format!("Its `with` does not read: {problem}")),

1322

},

1323

None => Ok(Map::new()),

1324

};

1325

let inputs = given.and_then(|given| crate::reach::call_inputs(&file, &trigger.inputs, &given));

1326

let secrets = crate::reach::secrets_plan(&job.raw, &contexts, outer.clone());

1327

let problem = match &inputs {

1328

Err(why) => Some(why.clone()),

1329

Ok(_) => crate::reach::missing_secrets(&called.raw, &secrets)

1330

.first()

1331

.map(|name| format!("`{file}` needs the secret `{name}`: pass it under `secrets:`, or use `secrets: inherit`."))

1332

.or_else(|| {

1333

crate::reach::undeclared_secrets(&called.raw, &secrets)

1334

.first()

1335

.map(|name| format!("Invalid secret: `{name}` is not defined in `{file}`."))

1336

}),

1337

};

1338

// Its own concurrency group, which it holds while its workflow runs.

let group = job

.concurrency

.as_ref()

.and_then(|c| expr::interpolate(&c.group, &scope).ok())

1343

.map(|group| group.trim().to_owned())

1344

.filter(|group| !group.is_empty());

let cancel = job

.concurrency

.as_ref()

.and_then(|c| expr::interpolate_value(&c.cancel_in_progress, &scope).ok())

1349

.is_some_and(|value| expr::truthy(&value));

1350

if problem.is_none()

1351

&& let Some(group) = &group

1352

&& !groups.iter().any(|(known, _)| known == group)

1353

{

1354

groups.push((group.clone(), cancel));

1355

}

1356

let call = json!({

1357

"role": "caller", "path": file, "source": source, "origin": origin, "depth": depth,

1358

"prefix": call_prefix(&row.key, job.matrix.is_some().then_some(index)), "title": name,

1359

"inputs": inputs.unwrap_or_default(), "secrets": secrets, "fail_fast": job.fail_fast,

1360

});

1361

let (status, conclusion, finished) = match &problem {

1362

Some(_) => ("completed", Some("failure"), Some(now())),

1363

None => ("waiting", None, None),

1364

};

1365

let values: Vec<worker::wasm_bindgen::JsValue> = vec![

1366

name.as_str().into(),

1367

serde_json::to_string(combination)?.into(),

1368

status.into(),

1369

optional(conclusion),

1370

optional(problem.as_deref()),

1371

job.max_parallel.map(|max| max.max(1)).map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),

1372

optional(finished.as_deref()),

1373

optional(group.as_deref()),

1374

u32::from(cancel).into(),

1375

serde_json::to_string(&call)?.into(),

1376

];

1377

if index == 0 {

1378

let mut bound = values;

1379

bound.push(row.id.as_str().into());

statements.push(

self.db

.prepare(

"UPDATE jobs SET name = ?, matrix = ?, status = ?, conclusion = ?, reason = ?, max_parallel = ?, finished_at = ?,

1384

concurrency_group = ?, cancel_in_progress = ?, call = ? WHERE id = ?",

)

.bind(&bound)?,

);

} else {

let mut bound: Vec<worker::wasm_bindgen::JsValue> = vec![

1390

new_id("job", now_ms()).into(),

1391

row.run_id.as_str().into(),

1392

row.repo_id.as_str().into(),

1393

row.namespace.as_str().into(),

1394

row.key.as_str().into(),

1395

(index as u32).into(),

1396

row.needs.as_str().into(),

1397

];

1398

bound.extend(values);

statements.push(

self.db

.prepare(

"INSERT INTO jobs (id, run_id, repo_id, namespace, key, ordinal, needs, name, matrix, status, conclusion, reason,

1403

max_parallel, finished_at, concurrency_group, cancel_in_progress, call)

1404

VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",

)

.bind(&bound)?,

);

}

}

self.db.batch(statements).await?;

1411

self.replace_in_groups(run, &row.key, &groups).await

1412

}

1413

1414

/// Whether another job of `row`'s concurrency group runs, or calls a

1415

/// workflow, so that `row` waits for it.

1416

async fn group_busy(&self, row: &JobRow) -> Result<bool> {

1417

let Some(group) = &row.concurrency_group else { return Ok(false) };

1418

let busy = self

1419

.db

1420

.prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?")

1421

.bind(&[row.repo_id.as_str().into(), group.as_str().into(), row.id.as_str().into()])?

1422

.first::<Count>(None)

1423

.await?

1424

.map_or(0, |count| count.n);

Ok(busy > 0)

}

/// A call that waited (`holds_call`) is made: the called workflow's

1429

/// jobs join the run under it, named under its name and keyed under its

1430

/// prefix. False when it no longer waits.

1431

async fn start_call(&self, row: &JobRow) -> Result<bool> {

1432

let Some(call) = row.call() else { return Ok(false) };

1433

let Some(prefix) = row.call_prefix() else { return Ok(false) };

1434

let Some(called) = call["source"].as_str().and_then(|source| workflow::parse(source).ok()) else { return Ok(false) };

1435

let file = call["path"].as_str().unwrap_or_default();

1436

let started = self

1437

.db

1438

.prepare("UPDATE jobs SET status = 'calling', reason = ?, started_at = ? WHERE id = ? AND status = 'waiting' RETURNING id")

1439

.bind(&[format!("Calls `{file}`.").into(), now().into(), row.id.as_str().into()])?

1440

.first::<Value>(None)

1441

.await?;

1442

if started.is_none() {

1443

return Ok(false);

1444

}

1445

let mut statements = Vec::new();

1446

for called_job in &called.jobs {

1447

let needs: Vec<String> = called_job.needs.iter().map(|n| format!("{prefix}/{n}")).collect();

1448

let callee = json!({

1449

"role": "callee", "parent": prefix, "title": row.name, "job": called_job.id, "path": file,

1450

"source": call["source"], "inputs": call["inputs"], "depth": call["depth"], "origin": call["origin"], "secrets": call["secrets"],

});

statements.push(

self.db

.prepare("INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, needs, status, call) VALUES (?, ?, ?, ?, ?, ?, ?, 'waiting', ?)")

1455

.bind(&[

1456

new_id("job", now_ms()).into(),

1457

row.run_id.as_str().into(),

1458

row.repo_id.as_str().into(),

1459

row.namespace.as_str().into(),

1460

format!("{prefix}/{}", called_job.id).into(),

1461

format!("{} / {}", row.name, called_job.name.clone().unwrap_or(called_job.id.clone())).into(),

1462

serde_json::to_string(&needs)?.into(),

1463

serde_json::to_string(&callee)?.into(),

])?,

);

}

if !statements.is_empty() {

1468

self.db.batch(statements).await?;

}

Ok(true)

}

/// The runs with a call waiting in `group`, but `run_id`, moved along

1474

/// now that a job of the group has finished.

1475

async fn wake_group(&self, repo_id: &str, group: &str, run_id: &str) -> Result<()> {

1476

#[derive(Deserialize)]

struct RunId {

run_id: String,

}

let waiting = self

.db

.prepare("SELECT DISTINCT run_id FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status = 'waiting' AND run_id != ?")

1483

.bind(&[repo_id.into(), group.into(), run_id.into()])?

1484

.all()

1485

.await?

1486

.results::<RunId>()?;

1487

for other in waiting {

1488

Box::pin(self.advance(&other.run_id)).await?;

}

Ok(())

}

/// A job that called a workflow, finished with its jobs: their result,

1494

/// and the outputs the workflow declares. A failed call of a fail-fast

1495

/// matrix stops the matrix's other calls.

1496

async fn finish_call(&self, run: &RunRow, row: &JobRow, children: &[&JobRow]) -> Result<()> {

1497

let call = row.call().unwrap_or_default();

1498

let called = call["source"].as_str().and_then(|s| workflow::parse(s).ok());

1499

let mut jobs_context = Map::new();

1500

let mut by_key: std::collections::BTreeMap<String, Vec<&JobRow>> = std::collections::BTreeMap::new();

1501

for child in children {

1502

by_key.entry(child.key.clone()).or_default().push(child);

1503

}

1504

for (key, rows) in &by_key {

1505

let mut outputs = Map::new();

1506

for child in rows {

1507

if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&child.outputs) {

1508

outputs.extend(more);

1509

}

1510

}

1511

let id = key.rsplit('/').next().unwrap_or(key);

1512

jobs_context.insert(id.to_owned(), json!({ "result": key_result(rows), "outputs": outputs }));

1513

}

1514

let inputs = children.first().and_then(|c| c.call()).map(|c| c["inputs"].clone()).unwrap_or(json!({}));

1515

let mut contexts = Map::new();

1516

contexts.insert("jobs".into(), Value::Object(jobs_context));

1517

contexts.insert("inputs".into(), inputs);

1518

let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };

1519

let mut outputs = Map::new();

1520

if let Some(Value::Object(declared)) = called.as_ref().map(|w| {

1521

let on = w.raw.get("on").or_else(|| w.raw.get("true")).cloned().unwrap_or(Value::Null);

1522

on.get("workflow_call").and_then(|c| c.get("outputs")).cloned().unwrap_or(Value::Null)

1523

}) {

1524

for (name, spec) in declared {

1525

if let Some(value) = spec.get("value") {

1526

let value = expr::interpolate_value(value, &scope).unwrap_or(Value::Null);

1527

outputs.insert(name, Value::String(expr::to_text(&value)));

}

}

}

let conclusion = key_result(children);

1532

let finished = self

1533

.db

1534

.prepare("UPDATE jobs SET status = 'completed', conclusion = ?, outputs = ?, finished_at = ? WHERE id = ? AND status = 'calling' RETURNING id")

1535

.bind(&[conclusion.into(), serde_json::to_string(&outputs)?.into(), now().into(), row.id.as_str().into()])?

1536

.first::<Value>(None)

1537

.await?;

1538

if finished.is_none() {

1539

return Ok(());

1540

}

1541

if conclusion == "failure" && row.matrix.as_deref().is_some_and(|m| m != "{}") && call["fail_fast"] != json!(false) {

1542

let siblings = self

1543

.db

1544

.prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")

1545

.bind(&[row.run_id.as_str().into(), row.key.as_str().into()])?

1546

.all()

1547

.await?

1548

.results::<JobRow>()?;

1549

for sibling in siblings {

1550

self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;

1551

}

1552

}

1553

if let Some(group) = &row.concurrency_group {

1554

self.wake_group(&run.repo_id, group, &run.id).await?;

}

Ok(())

}

/// A file's text at a commit, if it is there.

1560

pub(crate) async fn read_file(&self, path: &RepoPath, ws: &g1t_contracts::User, sha: &str, file: &str) -> Result<Option<String>> {

1561

let blob: Outcome<g1t_contracts::repos::BlobView> = g1t_kit::call(

1562

&self.repos,

1563

"blob",

1564

&g1t_contracts::repos::BlobArgs {

1565

path: path.clone(),

1566

viewer: Some(ws.clone()),

1567

git_ref: sha.to_owned(),

1568

file_path: file.to_owned(),

},

)

.await?;

Ok(match blob {

Outcome::Ok(view) => view.text,

1574

Outcome::Fail(_) => None,

})

}

async fn skip_job(&self, row: &JobRow, reason: Option<&str>) -> Result<()> {

1579

self.db

1580

.prepare("UPDATE jobs SET status = 'completed', conclusion = 'skipped', reason = ?, finished_at = ? WHERE id = ?")

1581

.bind(&[optional(reason), now().into(), row.id.as_str().into()])?

.run()

.await?;

Ok(())

}

async fn fail_job(&self, row: &JobRow, reason: &str) -> Result<()> {

1588

self.db

1589

.prepare("UPDATE jobs SET status = 'completed', conclusion = 'failure', reason = ?, finished_at = ? WHERE id = ? AND status != 'completed'")

1590

.bind(&[reason.into(), now().into(), row.id.as_str().into()])?

.run()

.await?;

Ok(())

}

/// Starts queued jobs, oldest first, while their workspace has room.

1597

pub async fn start_queued(&self) -> Result<()> {

1598

let queued = self

1599

.db

1600

// Self-hosted jobs are taken by their runners (runners.rs).

1601

.prepare("SELECT * FROM jobs WHERE status = 'queued' AND labels IS NULL ORDER BY rowid LIMIT 50")

1602

.all()

1603

.await?

1604

.results::<JobRow>()?;

1605

let mut running: std::collections::HashMap<String, u32> = std::collections::HashMap::new();

1606

// Every workspace's jobs on g1t's own machines: past its share, a

1607

// workspace starts more only while the pool has room (has_room).

1608

let mut in_pool = if queued.is_empty() {

0

} else {

self.db

.prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND runner_id IS NULL")

1613

.first::<Count>(None)

1614

.await?

1615

.map_or(0, |count| count.n)

1616

};

1617

for job in queued {

1618

let in_workspace = match running.get(&job.namespace) {

Some(n) => *n,

None => {

let n = self

.db

// Only g1t's own sandboxes count against the workspace's room.

1624

.prepare("SELECT COUNT(*) AS n FROM jobs WHERE status = 'in_progress' AND namespace = ? AND runner_id IS NULL")

1625

.bind(&[job.namespace.as_str().into()])?

1626

.first::<Count>(None)

1627

.await?

1628

.map_or(0, |count| count.n);

1629

running.insert(job.namespace.clone(), n);

n

}

};

if !has_room(in_workspace, in_pool) {

1634

continue;

1635

}

1636

if let Some(max) = job.max_parallel {

1637

let siblings = self

1638

.db

1639

.prepare("SELECT COUNT(*) AS n FROM jobs WHERE run_id = ? AND key = ? AND status = 'in_progress'")

1640

.bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?

1641

.first::<Count>(None)

1642

.await?

1643

.map_or(0, |count| count.n);

if siblings >= max {

continue;

}

}

// One job of a concurrency group runs at a time.

1649

if let Some(group) = &job.concurrency_group {

1650

let running = self

1651

.db

1652

.prepare("SELECT COUNT(*) AS n FROM jobs WHERE repo_id = ? AND concurrency_group = ? AND status IN ('in_progress', 'calling') AND id != ?")

1653

.bind(&[job.repo_id.as_str().into(), group.as_str().into(), job.id.as_str().into()])?

1654

.first::<Count>(None)

1655

.await?

1656

.map_or(0, |count| count.n);

if running > 0 {

continue;

}

}

let token = random_hex(24);

let at = now();

let claimed = self

.db

.prepare(

"UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",

1667

)

1668

.bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?

1669

.first::<Value>(None)

1670

.await?;

1671

if claimed.is_none() {

1672

continue;

1673

}

1674

running.insert(job.namespace.clone(), in_workspace + 1);

1675

in_pool += 1;

1676

self.db

1677

.prepare("UPDATE runs SET status = 'in_progress', started_at = COALESCE(started_at, ?) WHERE id = ? AND status = 'queued'")

1678

.bind(&[at.as_str().into(), job.run_id.as_str().into()])?

1679

.run()

1680

.await?;

1681

let run = self.run_row(&job.run_id).await?;

1682

let repo: RepoPath = run.as_ref().map(|run| repo_path(&run.repo)).unwrap_or(RepoPath {

1683

namespace: job.namespace.clone(),

1684

1685

});

1686

// The machine it asked for, for the runner.

1687

let instance = run.as_ref().and_then(|run| start_instance(run, &job));

1688

let trusted = run.as_ref().is_some_and(|run| run.trusted != 0);

1689

// Only a trusted run's jobs get secrets, brokered ones included.

1690

let brokered = trusted && self.may_broker(&job.repo_id, &job.namespace).await?;

1691

let started: Outcome<Value> = g1t_kit::call(

&self.runner,

"start_actions_job",

&StartJobArgs {

job: job.id.clone(),

token,

repo,

timeout_minutes: job.timeout_minutes,

instance,

brokered,

},

)

.await

.unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));

1705

if let Outcome::Fail(refused) = started {

1706

// Billing would not pay for a scheduled run's job: the

1707

// schedule waits rather than making runs only to refuse them.

1708

if refused.code == FailureCode::PaymentRequired

1709

&& let Some(run) = run.as_ref().filter(|run| run.event == "schedule")

1710

{

1711

self.db

1712

.prepare("UPDATE workflows SET schedule_refused_until = ? WHERE id = ?")

1713

.bind(&[rfc3339(now_ms() + crate::SCHEDULE_REFUSED_MS).into(), run.workflow_id.as_str().into()])?

.run()

.await?;

}

Box::pin(self.finish_job(&job.id, "failure", Some(&refused.message), None)).await?;

1718

} else {

1719

self.job_started(&job.id).await?;

}

}

Ok(())

}

/// Finishes a job and moves its run along.

1726

pub async fn finish_job(&self, job_id: &str, conclusion: &str, reason: Option<&str>, outputs: Option<&Map<String, Value>>) -> Result<()> {

let finished = self

.db

.prepare(

"UPDATE jobs SET status = 'completed', conclusion = ?, reason = COALESCE(?, reason), outputs = COALESCE(?, outputs),

1731

finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *",

)

.bind(&[

conclusion.into(),

optional(reason),

outputs.map(|o| serde_json::to_string(o).unwrap_or_default()).as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),

now().into(),

job_id.into(),

])?

.first::<JobRow>(None)

1741

.await?;

1742

let Some(job) = finished else { return Ok(()) };

1743

self.tell_live(&job.repo_id, &job.run_id, None, None).await;

1744

// Its G1T_TOKEN stops working with it.

1745

self.revoke_job_tokens(&job.id).await;

1746

// A g1t.page build it opened and did not finish fails now, not at

1747

// the deployments sweep. Only a job that ran can have opened one.

1748

if job.started_at.is_some() {

1749

self.end_job_builds(&job.id, conclusion).await;

1750

}

1751

// A job that deploys failed: so did its run's deployment, now.

1752

if conclusion == "failure" && job.continue_on_error == 0 && job.started_at.is_some()

1753

&& let Some(run) = self.run_row(&job.run_id).await?

1754

&& let Some(env) = deploys_to(&run, &job)

1755

{

1756

self.report_deployment(&run, &env, "failure", false).await;

1757

}

1758

// A self-hosted runner's job: the runner is free again, and its time

1759

// is recorded, at nothing.

1760

if job.runner_id.is_some() {

1761

self.released(&job).await?;

1762

}

1763

// Steps still marked as going are not going any more.

1764

let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();

1765

let mut touched = false;

1766

for step in steps.iter_mut() {

1767

if step["status"] != "completed" {

1768

let was_running = step["status"] == "in_progress";

1769

step["status"] = json!("completed");

1770

step["conclusion"] = json!(if was_running { conclusion } else { "skipped" });

touched = true;

}

}

if touched {

self.db

.prepare("UPDATE jobs SET steps = ? WHERE id = ?")

1777

.bind(&[serde_json::to_string(&steps)?.into(), job.id.as_str().into()])?

.run()

.await?;

}

// fail-fast: one failed combination stops the rest of its matrix.

1782

if conclusion == "failure" && job.continue_on_error == 0 && job.matrix.as_deref().is_some_and(|m| m != "{}") {

1783

let run = self.run_row(&job.run_id).await?;

1784

// As its workflow, or the workflow it is a job of, says.

1785

let fail_fast = run.as_ref().and_then(|run| job_spec(run, &job)).is_none_or(|j| j.fail_fast);

if fail_fast {

let siblings = self

.db

.prepare("SELECT * FROM jobs WHERE run_id = ? AND key = ? AND status != 'completed'")

1790

.bind(&[job.run_id.as_str().into(), job.key.as_str().into()])?

1791

.all()

1792

.await?

1793

.results::<JobRow>()?;

1794

for sibling in siblings {

1795

self.stop_job(&sibling, "Another job of its matrix failed, and the matrix is fail-fast.").await?;

}

}

}

// A call waiting in its concurrency group, in another run, may start.

1800

if let Some(group) = &job.concurrency_group {

1801

self.wake_group(&job.repo_id, group, &job.run_id).await?;

1802

}

1803

Box::pin(self.advance(&job.run_id)).await

1804

}

1805

1806

/// Cancels a job. One that is running is told to stop (the answer to

1807

/// its next report), and runs its `if: always()` and `cancelled()`

1808

/// steps and its post steps before it ends `cancelled`; one that has

1809

/// not started is cancelled at once.

1810

async fn stop_job(&self, job: &JobRow, reason: &str) -> Result<()> {

1811

// A job calling a workflow stops with that workflow's jobs.

1812

if job.status == "calling"

1813

&& let Some(prefix) = job.call_prefix()

1814

{

1815

let under = format!("{prefix}/");

1816

for child in self.job_rows(&job.run_id).await?.iter().filter(|child| child.key.starts_with(&under) && child.status != "completed") {

1817

Box::pin(self.stop_job(child, reason)).await?;

1818

}

1819

}

1820

if job.status == "in_progress" && job.started_at.is_some() {

1821

self.db

1822

.prepare("UPDATE jobs SET cancel_requested_at = COALESCE(cancel_requested_at, ?), reason = ? WHERE id = ? AND status = 'in_progress'")

1823

.bind(&[now().into(), reason.into(), job.id.as_str().into()])?

.run()

.await?;

return Ok(());

}

self.hard_stop(job, reason).await

1829

}

1830

1831

/// Cancels a job outright, stopping its sandbox if it has one.

1832

async fn hard_stop(&self, job: &JobRow, reason: &str) -> Result<()> {

1833

// A self-hosted runner hears it was cancelled on its next poll.

1834

if job.status == "in_progress" && job.runner_id.is_none() {

1835

let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;

}

let stopped = self

.db

.prepare("UPDATE jobs SET status = 'completed', conclusion = 'cancelled', reason = ?, finished_at = ?, token_hash = NULL, first_token_hash = NULL WHERE id = ? AND status != 'completed' RETURNING *")

1840

.bind(&[reason.into(), now().into(), job.id.as_str().into()])?

1841

.first::<JobRow>(None)

1842

.await?;

1843

if stopped.as_ref().is_some_and(|row| row.started_at.is_some()) {

1844

self.revoke_job_tokens(&job.id).await;

1845

self.end_job_builds(&job.id, "cancelled").await;

1846

}

1847

if let Some(stopped) = stopped.filter(|row| row.runner_id.is_some()) {

1848

self.released(&stopped).await?;

}

Ok(())

}

/// Ends a job's tokens at once. A failure is logged: the token expires

1854

/// on its own soon after the job's time limit.

1855

async fn revoke_job_tokens(&self, job_id: &str) {

1856

let revoked: Result<bool> =

1857

g1t_kit::call(&self.identity, "revoke_job_tokens", &RevokeJobTokensArgs { job_id: job_id.to_owned() }).await;

1858

if let Err(error) = revoked {

1859

worker::console_error!("actions: the tokens of job {job_id} were not revoked: {error}");

}

}

/// Finishes the run when every job has.

1864

async fn finish_if_done(&self, run_id: &str) -> Result<()> {

1865

let Some(run) = self.run_row(run_id).await? else { return Ok(()) };

1866

if matches!(run.status.as_str(), "completed" | "pending" | "action_required") {

1867

return Ok(());

1868

}

1869

let jobs = self.job_rows(run_id).await?;

1870

if !jobs.iter().all(|job| job.status == "completed") {

1871

return Ok(());

1872

}

1873

self.finish_run(&run, None).await

1874

}

1875

1876

async fn finish_run(&self, run: &RunRow, error: Option<&str>) -> Result<()> {

1877

let jobs = self.job_rows(&run.id).await?;

1878

let rows: Vec<&JobRow> = jobs.iter().collect();

1879

let conclusion = if error.is_some() {

1880

"failure"

1881

} else if run.conclusion.as_deref() == Some("cancelled") {

1882

"cancelled"

1883

} else if rows.is_empty() {

"skipped"

} else {

key_result(&rows)

};

let done = self

.db

.prepare("UPDATE runs SET status = 'completed', conclusion = ?, error = COALESCE(?, error), finished_at = ? WHERE id = ? AND status != 'completed' RETURNING id")

1891

.bind(&[conclusion.into(), optional(error), now().into(), run.id.as_str().into()])?

1892

.first::<Value>(None)

.await?;

if done.is_none() {

return Ok(());

}

self.report_status(run, conclusion).await?;

1898

self.settle_deployments(run, &jobs, conclusion == "cancelled").await;

1899

let published: Result<()> = g1t_kit::call(

1900

&self.events,

1901

"publish",

1902

&g1t_contracts::events::Publish {

1903

events: vec![g1t_contracts::events::NewEvent {

1904

kind: "workflow.completed",

1905

source: "actions",

1906

repo_id: Some(run.repo_id.clone()),

1907

actor: run.actor_id.clone(),

1908

data: g1t_contracts::events::WorkflowEvent {

1909

run_id: run.id.clone(),

1910

repo_id: run.repo_id.clone(),

1911

workflow: run.name.clone(),

1912

path: run.path.clone(),

1913

number: run.number,

1914

event: run.event.clone(),

1915

conclusion: conclusion.to_owned(),

1916

git_ref: run.git_ref.clone(),

1917

sha: run.sha.clone(),

pull: run.pull,

},

}],

},

)

.await;

if let Err(error) = published {

1925

worker::console_error!("actions: could not publish workflow.completed: {error}");

1926

}

1927

// The next run waiting in its concurrency group.

1928

if let Some(group) = &run.concurrency_group {

1929

let next = self

1930

.db

1931

.prepare("SELECT * FROM runs WHERE repo_id = ? AND concurrency_group = ? AND status = 'pending' ORDER BY id LIMIT 1")

1932

.bind(&[run.repo_id.as_str().into(), group.as_str().into()])?

1933

.first::<RunRow>(None)

1934

.await?;

1935

if let Some(next) = next {

1936

self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[next.id.as_str().into()])?.run().await?;

1937

Box::pin(self.advance(&next.id)).await?;

}

}

Ok(())

}

/// Tells the deployments service that a job ended (`workflow_job_ended`),

1944

/// so a build it opened with `start_from_workflow` and left open fails

1945

/// at once. Never fails the job: an answer that does not come is

1946

/// logged, and the sweep fails the build later.

1947

async fn end_job_builds(&self, job_id: &str, conclusion: &str) {

1948

let ended: Result<Value> =

1949

g1t_kit::call(&self.deployments, "workflow_job_ended", &json!({ "jobId": job_id, "conclusion": conclusion })).await;

1950

if let Err(error) = ended {

1951

worker::console_log!("actions: could not tell deployments that job {job_id} ended: {error}");

}

}

/// Tells the deployments service how a run's deployment to `env` stands:

1956

/// made when its first job naming the environment starts, failed when one

1957

/// of them fails, and settled (`last`) when the run finishes. Never fails

1958

/// the run: a deployment that could not be recorded is logged.

1959

async fn report_deployment(&self, run: &RunRow, env: &JobEnvironment, state: &str, last: bool) {

1960

let path = repo_path(&run.repo);

1961

let reported: Result<Value> = g1t_kit::call(

1962

&self.deployments,

1963

"actions_deployment",

1964

&json!({

1965

"repoId": run.repo_id,

1966

"repo": { "namespace": path.namespace, "name": path.name },

1967

"runId": run.id,

1968

"attempt": run.attempt,

1969

"runUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),

1970

"environment": env.name,

"url": env.url,

"ref": run.git_ref,

"sha": run.sha,

"state": state,

"final": last,

"creator": run.actor,

1977

"workflow": run.name,

}),

)

.await;

if let Err(error) = reported {

1982

worker::console_error!("actions: deployment to {} not recorded for run {}: {error}", env.name, run.id);

}

}

/// A job that deploys has started: its run's deployment to the

1987

/// environment is under way.

1988

pub(crate) async fn job_started(&self, job_id: &str) -> Result<()> {

1989

let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[job_id.into()])?.first::<JobRow>(None).await? else {

1990

return Ok(());

1991

};

1992

let Some(run) = self.run_row(&job.run_id).await? else { return Ok(()) };

1993

self.tell_live(&run.repo_id, &run.id, Some(&run.sha), run.pull).await;

1994

if let Some(env) = deploys_to(&run, &job) {

1995

self.report_deployment(&run, &env, "in_progress", false).await;

}

Ok(())

}

/// Pages open on the repository hear that a run's job started or

2001

/// finished, which is no event (`g1t_contracts::live`). Logged and left

2002

/// when notify cannot be reached: the pages catch up by themselves.

2003

pub(crate) async fn tell_live(&self, repo_id: &str, run_id: &str, sha: Option<&str>, pull: Option<u32>) {

2004

let Some(notify) = &self.notify else { return };

2005

let change = g1t_contracts::live::RepoChange {

2006

repo_id: repo_id.to_owned(),

2007

kind: g1t_contracts::live::RUN_UPDATED.to_owned(),

2008

run: Some(run_id.to_owned()),

2009

sha: sha.map(str::to_owned),

number: pull,

at: now(),

..Default::default()

};

let args = g1t_contracts::live::RepoChangedArgs { changes: vec![change] };

2015

let told: Result<Value> = g1t_kit::call(notify, g1t_contracts::live::REPO_CHANGED, &args).await;

2016

if let Err(error) = told {

2017

worker::console_error!("actions: live pages not told of run {run_id}: {error}");

}

}

/// The run is over: each environment its jobs deployed to takes the

2022

/// outcome of those jobs (`deployment_outcome`).

2023

async fn settle_deployments(&self, run: &RunRow, jobs: &[JobRow], cancelled: bool) {

2024

let mut seen: Vec<(JobEnvironment, Vec<Option<String>>)> = Vec::new();

2025

for job in jobs {

2026

let Some(env) = deploys_to(run, job) else { continue };

2027

let conclusion = if cancelled && job.conclusion.as_deref() != Some("skipped") && job.started_at.is_some() {

2028

Some("cancelled".to_owned())

2029

} else if job.started_at.is_none() {

2030

Some("skipped".to_owned())

2031

} else {

2032

job.conclusion.clone()

2033

};

2034

match seen.iter_mut().find(|(known, _)| known.name.eq_ignore_ascii_case(&env.name)) {

2035

Some((known, conclusions)) => {

2036

if known.url.is_none() {

2037

known.url = env.url.clone();

2038

}

2039

conclusions.push(conclusion);

2040

}

2041

None => seen.push((env, vec![conclusion])),

2042

}

2043

}

2044

for (env, conclusions) in seen {

2045

if let Some(state) = deployment_outcome(&conclusions) {

2046

self.report_deployment(run, &env, state, true).await;

}

}

}

/// Tells the pull request (or commit) how the run went, as a status.

2052

async fn report_status(&self, run: &RunRow, conclusion: &str) -> Result<()> {

2053

let state = match conclusion {

2054

"success" | "skipped" => "success",

2055

"cancelled" => "error",

2056

_ => "failure",

2057

};

2058

let _: Result<Value> = g1t_kit::call(

&self.work,

"set_commit_status",

&json!({

"repoId": run.repo_id,

2063

"sha": run.sha,

2064

"context": format!("{} / {}", run.name, run.event),

2065

"state": state,

2066

"description": format!("{} {}", run.name, match conclusion {

2067

"success" => "passed",

2068

"skipped" => "was skipped",

2069

"cancelled" => "was cancelled",

2070

_ => "failed",

2071

}),

2072

"targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),

"source": "actions",

}),

)

.await;

Ok(())

}

/// Tells the pull request a run has started on its head.

2081

pub async fn report_pending(&self, run: &RunRow) -> Result<()> {

2082

if let Err(error) = self.try_report_pending(run).await {

2083

worker::console_warn!("pending status for run {} not reported: {error}", run.id);

}

Ok(())

}

/// Reports a run as pending on its commit, failing when Work did not

2089

/// store it.

2090

pub async fn try_report_pending(&self, run: &RunRow) -> Result<()> {

2091

let stored: Outcome<bool> = g1t_kit::call(

&self.work,

"set_commit_status",

&json!({

"repoId": run.repo_id,

2096

"sha": run.sha,

2097

"context": format!("{} / {}", run.name, run.event),

2098

"state": "pending",

2099

"description": if run.status == "action_required" {

2100

format!("{} is waiting for approval", run.name)

2101

} else {

2102

format!("{} is running", run.name)

2103

},

2104

"targetUrl": format!("{SITE}/{}/actions/runs/{}", run.repo, run.id),

"source": "actions",

}),

)

.await?;

match stored {

Outcome::Ok(_) => Ok(()),

2111

Outcome::Fail(failure) => Err(worker::Error::RustError(format!("pending status refused: {}", failure.message))),

}

}

/// Cancels a run: its waiting and queued jobs at once, and its running

2116

/// ones gracefully (`stop_job`), or outright when `force`.

2117

pub async fn cancel_run(&self, run: &RunRow, reason: &str, force: bool) -> Result<()> {

2118

self.db

2119

.prepare("UPDATE runs SET conclusion = 'cancelled' WHERE id = ? AND status != 'completed'")

2120

.bind(&[run.id.as_str().into()])?

2121

.run()

2122

.await?;

2123

for job in self.job_rows(&run.id).await?.iter().filter(|job| job.status != "completed") {

2124

if force {

2125

self.hard_stop(job, reason).await?;

2126

} else {

2127

self.stop_job(job, reason).await?;

2128

}

2129

}

2130

if run.status == "pending" || run.status == "action_required" {

2131

self.db.prepare("UPDATE runs SET status = 'queued' WHERE id = ?").bind(&[run.id.as_str().into()])?.run().await?;

2132

}

2133

self.advance(&run.id).await

2134

}

2135

2136

/// Cancels every run of the repository that has not finished, for

2137

/// `repo.deleted` and `repo.archived`.

2138

pub async fn stop_runs(&self, repo_id: &str) -> Result<()> {

2139

let runs = self

2140

.db

2141

.prepare("SELECT * FROM runs WHERE repo_id = ? AND status != 'completed'")

2142

.bind(&[repo_id.into()])?

2143

.all()

2144

.await?

2145

.results::<RunRow>()?;

2146

for run in runs {

2147

self.cancel_run(&run, "The repository was archived or deleted.", true).await?;

}

Ok(())

}

pub async fn cancel(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {

2153

if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {

2154

return Ok(Outcome::Fail(refused));

2155

}

2156

let run = check!(self.run_in(&a.repo, &a.id).await?);

2157

if run.status == "completed" {

2158

return Ok(fail(FailureCode::Conflict, "The run has already finished."));

2159

}

2160

// Cancelling a run that is already cancelling stops its jobs

2161

// outright, without waiting for their cleanup steps.

2162

let force = a.force || run.conclusion.as_deref() == Some("cancelled");

2163

let reason = if force {

2164

format!("{} stopped the run without waiting for its cleanup steps.", a.actor.username)

2165

} else {

2166

format!("{} cancelled the run.", a.actor.username)

2167

};

2168

self.cancel_run(&run, &reason, force).await?;

2169

self.run_summary(&run.id).await

2170

}

2171

2172

/// Runs again, as a new attempt: every job, with `failed_only` those

2173

/// that did not succeed, or with `job` that one; each with the jobs that

2174

/// need them. The attempt that ends is kept, its jobs and their logs and

2175

/// summaries with it (`job_attempts`, `run_attempts`).

2176

pub async fn rerun(&self, a: RunActionArgs) -> Result<Outcome<WorkflowRun>> {

2177

if let Outcome::Fail(refused) = self.may(&a.actor, &a.repo, Capability::Run).await? {

2178

return Ok(Outcome::Fail(refused));

2179

}

2180

// A job alone (`POST …/jobs/{job}/rerun`) names its run.

2181

let run_id = match (&a.job, a.id.is_empty()) {

2182

(Some(job), true) => {

2183

#[derive(Deserialize)]

struct Of {

run_id: String,

}

let of = self.db.prepare("SELECT run_id FROM jobs WHERE id = ?").bind(&[job.as_str().into()])?.first::<Of>(None).await?;

2188

match of {

2189

Some(of) => of.run_id,

2190

None => return Ok(fail(FailureCode::NotFound, "No such job.")),

}

}

_ => a.id.clone(),

};

let run = check!(self.run_in(&a.repo, &run_id).await?);

2196

if run.status != "completed" {

2197

return Ok(fail(FailureCode::Conflict, "The run is still going: cancel it first."));

2198

}

2199

// Nothing starts again on an archived repository.

2200

match self.visible_repo(&a.repo, &Some(a.actor.clone())).await? {

2201

Some(repo) if repo.archived() => {

2202

return Ok(fail(FailureCode::Forbidden, g1t_contracts::repos::archived_message(&repo.namespace, &repo.name)));

2203

}

2204

Some(repo) if !self.actions_on(&repo).await? => {

2205

return Ok(fail(FailureCode::Forbidden, g1t_contracts::actions::actions_disabled_message(&repo.namespace, &repo.name)));

2206

}

2207

Some(_) => {}

2208

None => return Ok(fail(FailureCode::NotFound, "There is no such repository.")),

2209

}

2210

if run.error.is_some() {

2211

return Ok(fail(FailureCode::Conflict, "This run never started: fix the workflow file and push again."));

2212

}

2213

let jobs = self.job_rows(&run.id).await?;

2214

let workflow = workflow::parse(&run.source).ok();

2215

let target = match &a.job {

2216

Some(id) => match jobs.iter().find(|job| &job.id == id) {

2217

Some(job) => Some(top_key(&job.key).to_owned()),

2218

None => return Ok(fail(FailureCode::NotFound, "That job is not in the run's latest attempt.")),

},

None => None,

};

let which = match (&target, a.failed_only) {

2223

(Some(key), _) => Rerun::Job(key),

2224

(None, true) => Rerun::Failed,

2225

(None, false) => Rerun::All,

2226

};

2227

let order = workflow.as_ref().map(|w| w.job_order()).unwrap_or_default();

2228

let again = rerun_keys(

2229

&order,

2230

|key| jobs.iter().find(|j| j.key == key).map(JobRow::needs).unwrap_or_default(),

2231

|key| jobs.iter().filter(|j| j.key == key).all(|j| j.conclusion.as_deref() == Some("success")),

2232

which,

2233

);

2234

if again.is_empty() {

2235

return Ok(fail(FailureCode::Conflict, "Every job succeeded: there is nothing to run again."));

2236

}

2237

// The jobs that run again, a called workflow's with the job calling it.

2238

let rerun_ids: Vec<&str> = jobs.iter().filter(|job| again.iter().any(|key| key == top_key(&job.key))).map(|job| job.id.as_str()).collect();

2239

let ids = serde_json::to_string(&rerun_ids)?;

2240

let ended = run.attempt.to_string();

2241

let ended = ended.as_str();

2242

let mut statements = vec![

2243

// The attempt that ends, as it ended.

2244

self.db

2245

.prepare(

2246

"INSERT OR REPLACE INTO run_attempts (run_id, attempt, repo_id, conclusion, actor, debug, started_at, finished_at)

2247

SELECT id, attempt, repo_id, conclusion, COALESCE(triggering_actor, actor), debug, started_at, finished_at FROM runs WHERE id = ?",

2248

)

2249

.bind(&[run.id.as_str().into()])?,

2250

// Earlier attempts that showed a job's logs from where it ran

2251

// then now find them where they move to.

2252

self.db

2253

.prepare("UPDATE job_attempts SET log_id = log_id || '.' || ?1 WHERE run_id = ?2 AND log_id IN (SELECT value FROM json_each(?3))")

2254

.bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,

2255

// Each of its jobs: one that runs again keeps its logs under

2256

// `{id}.{attempt}`; one left alone is still the live job's.

2257

self.db

2258

.prepare(

2259

"INSERT OR REPLACE INTO job_attempts (id, run_id, repo_id, attempt, job_id, log_id, key, ordinal, name, needs, status, conclusion,

2260

steps, annotations, reason, environment, labels, runner_name, started_at, finished_at)

2261

SELECT id || '.' || ?1, run_id, repo_id, ?1, id,

2262

CASE WHEN id IN (SELECT value FROM json_each(?3)) THEN id || '.' || ?1 ELSE id END,

2263

key, ordinal, name, needs, status, conclusion, steps, annotations, reason, environment, labels, runner_name, started_at, finished_at

2264

FROM jobs WHERE run_id = ?2 ORDER BY rowid",

2265

)

2266

.bind(&[ended.into(), run.id.as_str().into(), ids.as_str().into()])?,

2267

self.db

2268

.prepare("UPDATE logs SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")

2269

.bind(&[ended.into(), ids.as_str().into()])?,

2270

self.db

2271

.prepare("UPDATE job_summaries SET job_id = job_id || '.' || ?1 WHERE job_id IN (SELECT value FROM json_each(?2))")

2272

.bind(&[ended.into(), ids.as_str().into()])?,

2273

];

2274

for key in &again {

2275

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()])?);

2276

// The jobs of a workflow it called (each matrix combination's

2277

// under `key[i]/`) are made again when it calls it again.

2278

statements.push(

2279

self.db

2280

.prepare("DELETE FROM jobs WHERE run_id = ? AND (key LIKE ? OR key LIKE ?)")

2281

.bind(&[run.id.as_str().into(), format!("{key}/%").into(), format!("{key}[%").into()])?,

);

statements.push(

self.db

.prepare(

"UPDATE jobs SET status = 'waiting', conclusion = NULL, steps = '[]', annotations = '[]', outputs = '{}', reason = NULL,

2287

matrix = NULL, call = NULL, token_hash = NULL, first_token_hash = NULL, seen_at = NULL, started_at = NULL, finished_at = NULL,

2288

labels = NULL, queued_at = NULL, runner_id = NULL, runner_name = NULL, environment = NULL,

2289

concurrency_group = NULL, cancel_in_progress = 0, cancel_requested_at = NULL WHERE run_id = ? AND key = ?",

2290

)

2291

.bind(&[run.id.as_str().into(), key.as_str().into()])?,

);

}

statements.push(

self.db

.prepare(

"UPDATE runs SET status = 'queued', conclusion = NULL, attempt = attempt + 1, started_at = NULL, finished_at = NULL,

2298

triggering_actor = ?, debug = ? WHERE id = ?",

2299

)

2300

.bind(&[a.actor.username.as_str().into(), u32::from(a.debug).into(), run.id.as_str().into()])?,

2301

);

2302

self.db.batch(statements).await?;

2303

if let Some(run) = self.run_row(&run.id).await? {

2304

self.report_pending(&run).await?;

2305

}

2306

self.advance(&run.id).await?;

2307

// A page left open on the attempt that ended, which asks for nothing

2308

// while its run is over, hears that it is going again: re-run from

2309

// the API, the MCP tool, a pull request or another tab, as well as

2310

// from that page, it offers Cancel run and not the re-runs.

2311

self.tell_live(&run.repo_id, &run.id, Some(&run.sha), run.pull).await;

2312

self.run_summary(&run.id).await

2313

}

2314

2315

pub async fn run_in(&self, repo: &RepoPath, id: &str) -> Result<Outcome<RunRow>> {

2316

let row = self

2317

.db

2318

.prepare("SELECT * FROM runs WHERE id = ? AND lower(repo) = lower(?)")

2319

.bind(&[id.into(), format!("{}/{}", repo.namespace, repo.name).into()])?

2320

.first::<RunRow>(None)

2321

.await?;

2322

Ok(row.map_or_else(|| fail(FailureCode::NotFound, "No such run."), Outcome::Ok))

2323

}

2324

2325

// --- The sandbox's side -----------------------------------------------------

2326

2327

pub(crate) async fn job_for_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {

2328

self.job_for(a, false).await

2329

}

2330

2331

/// As [`Self::job_for_token`], and also with the token the job started

2332

/// with after its harness swapped it (`job_spec` with `rotate`): for

2333

/// `brokered_secrets` alone, which the runner Worker asks with the

2334

/// token it was given at the start, and which no API route passes on.

2335

pub(crate) async fn job_for_started_token(&self, a: &JobCallArgs) -> Result<Outcome<JobRow>> {

2336

self.job_for(a, true).await

2337

}

2338

2339

async fn job_for(&self, a: &JobCallArgs, started: bool) -> Result<Outcome<JobRow>> {

2340

let job = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[a.job.as_str().into()])?.first::<JobRow>(None).await?;

2341

Ok(match job {

2342

Some(job) if job.status == "in_progress" && token_opens(job.token_hash.as_deref(), job.first_token_hash.as_deref(), &a.token, started) => Outcome::Ok(job),

2343

_ => fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."),

})

}

/// `job_auth`: which run and repository a running job's token is for,

2348

/// so the API can keep its artifacts and cache.

2349

pub async fn job_auth(&self, a: JobCallArgs) -> Result<Outcome<Value>> {

2350

let job = check!(self.job_for_token(&a).await?);

2351

Ok(Outcome::Ok(json!({ "run": job.run_id, "repoId": job.repo_id })))

2352

}

2353

2354

/// `job_spec`: everything the sandbox needs to run the job.

2355

pub async fn job_spec(&self, a: JobCallArgs) -> Result<Outcome<Value>> {

2356

let job = check!(self.job_for_token(&a).await?);

2357

// Asked to, the token the harness asked with is spent here: it gets

2358

// a new one with the spec, for everything after, so a copy of the

2359

// first left on the runner's computer (a Windows job's script, a

2360

// container's environment) cannot fetch the secrets again. The new

2361

// token is made now, as the runtime token is signed with its hash,

2362

// and swapped in last, so a spec that could not be made leaves the

2363

// old one working.

2364

let next = a.rotate.then(|| random_hex(24));

2365

let token_hash = match &next {

2366

Some(next) => sha256_hex(next),

2367

None => job.token_hash.clone().unwrap_or_default(),

2368

};

2369

let Some(run) = self.run_row(&job.run_id).await? else {

2370

return Ok(fail(FailureCode::NotFound, "No such run."));

2371

};

2372

let Ok(caller) = workflow::parse(&run.source) else {

2373

return Ok(fail(FailureCode::Invalid, "The workflow no longer reads."));

2374

};

2375

// A called workflow's job runs as that workflow defines it.

2376

let callee = job.callee();

2377

let (workflow, spec, call_inputs) = match callee {

2378

Some((called, spec, call)) => (called, spec, Some(call["inputs"].clone())),

2379

None => match caller.jobs.iter().find(|j| j.id == job.key) {

2380

Some(spec) => (caller.clone(), spec.clone(), None),

2381

None => return Ok(fail(FailureCode::NotFound, "The job is not in the workflow.")),

},

};

let spec = &spec;

let repo = repo_path(&run.repo);

2386

let trusted = run.trusted != 0;

2387

// The job's `environment:`, by name, as read when its needs were done

2388

// (an expression included), once the environment's protection rules

2389

// let it start: entries with a value for it give that value instead

2390

// of their default, as GitHub's environment secrets do.

2391

let environment: Option<String> = job.environment.clone();

2392

// What its token may do: its `permissions:` (a called workflow's

2393

// jobs no more than the job that calls it), else the repository's

2394

// default; read-only for a pull request from outside.

2395

// The repository's default, its workspace's taken in, and whether

2396

// its jobs may open and approve pull requests.

2397

let (default, pull_requests) = self.token_policy(&run.repo_id).await?;

2398

let mut permissions = spec.permissions(&workflow, default);

2399

if let Some(parent) = &job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned)) {

2400

let top = top_key(parent);

2401

if let Some(caller_job) = caller.jobs.iter().find(|j| j.id == top) {

2402

permissions = permissions.capped_by(&caller_job.permissions(&caller, default));

}

}

if !trusted {

permissions = permissions.read_only();

2407

}

2408

// G1T_TOKEN, and GITHUB_TOKEN as its alias: a token of the

2409

// workspace's that reaches this repository only, with the scopes

2410

// its permissions give, until the job ends.

2411

let token = match self.workspace_actor(&repo.namespace).await? {

2412

Some(workspace) => {

2413

let created: CreatedAccessToken = g1t_kit::call(

2414

&self.identity,

2415

"create_job_token",

2416

&CreateJobTokenArgs {

2417

workspace,

2418

repo: repo.clone(),

2419

run_id: run.id.clone(),

2420

job_id: job.id.clone(),

2421

2422

ttl_seconds: u64::from(job.timeout_minutes) * 60 + 600,

2423

scopes: permissions.scopes().into_iter().map(str::to_owned).collect(),

2424

pull_requests: pull_requests && trusted,

},

)

.await?;

created.token

}

None => String::new(),

2431

};

2432

// A run that is not trusted (a pull request from outside the

2433

// workspace) gets no secrets and an empty token.

2434

let passed = job.call().filter(|c| c["role"] == "callee").and_then(|c| c.get("secrets").cloned());

2435

// Brokered secrets among them (broker.rs), whatever keys they reach

2436

// the job under.

2437

let mut brokers: Vec<Broker> = Vec::new();

2438

let mut secrets = match (trusted, passed) {

2439

(false, _) => Map::new(),

2440

(true, None) => {

2441

let resolved = self.secrets_for(&run.repo_id, &run.repo, environment.as_deref(), true).await?;

2442

brokers.extend(resolved.brokers());

2443

resolved.values

2444

}

2445

// A called workflow's job: what its callers passed it (reach.rs),

2446

// and its own environment's secrets over them.

2447

(true, Some(plan)) => {

2448

let base = self.secrets_for(&run.repo_id, &run.repo, None, true).await?;

2449

brokers.extend(base.brokers());

2450

let base = base.values;

2451

let vars = self.variables_for(&run.repo_id, &run.repo, None, true).await?;

2452

let github = run.info().context(&job.key, "", run.action.as_deref());

2453

let mut passed = crate::reach::resolve_secrets(&plan, &base, &github, &vars);

2454

if let Some(name) = environment.as_deref() {

2455

let own = self.secrets_for(&run.repo_id, &run.repo, Some(name), true).await?;

2456

brokers.extend(own.brokers());

2457

for (key, value) in own.values {

2458

if base.get(&key) != Some(&value) {

2459

passed.insert(key, value);

}

}

}

passed

}

};

// A brokered secret's value never reaches the sandbox: the job gets

2467

// a stand-in, which the runner Worker fills in on its requests to

2468

// the secret's hosts. A self-hosted runner's requests never pass

2469

// the Worker: a harness there that brokers is given each stand-in's

2470

// hosts, and passes its job's requests to them through g1t

2471

// (broker.rs); any other keeps a stand-in that does nothing, and

2472

// the log says so.

2473

let stand_ins = broker::substitute(&mut secrets, &brokers, broker::stand_in);

2474

let relayed = job.runner_id.is_some() && a.broker && !stand_ins.is_empty();

2475

if !stand_ins.is_empty() {

2476

self.keep_stand_ins(&job.id, job.timeout_minutes, &stand_ins).await?;

2477

if job.runner_id.is_some() {

2478

let mut made: Vec<&broker::StandIn> = stand_ins.iter().collect();

2479

made.sort_unstable_by(|a, b| a.broker.name.cmp(&b.broker.name));

2480

let notes: String = made

2481

.iter()

2482

.map(|made| {

2483

let note = if relayed { broker::brokered_note(&made.broker.name, &made.broker.hosts) } else { broker::self_hosted_note(&made.broker.name) };

note + "\n"

})

.collect();

self.note_log(&job.id, 0, &notes).await?;

2488

}

2489

}

2490

secrets.insert("G1T_TOKEN".into(), Value::String(token.clone()));

2491

secrets.insert("GITHUB_TOKEN".into(), Value::String(token.clone()));

2492

// Each secret as it is, a line at a time, base64 and JSON-escaped.

2493

let mut masks: Vec<String> = g1t_actions::mask::all_variants(secrets.values().filter_map(|v| v.as_str()));

2494

let vars = self.variables_for(&run.repo_id, &run.repo, environment.as_deref(), trusted).await?;

2495

// The toolkit's runtime token (runtime.rs), for as long as the job

2496

// may run; the API puts it and the toolkit's addresses in the

2497

// job's variables.

2498

let runtime_token = crate::runtime::runtime_token(

&job.id,

&job.run_id,

&token_hash,

now_ms() / 1000,

u64::from(job.timeout_minutes) * 60 + 600,

2504

);

2505

masks.push(runtime_token.clone());

2506

let retention_days = self.retention_setting(&run.repo_id).await?;

2507

2508

let jobs = self.job_rows(&run.id).await?;

2509

let mut needs = Map::new();

2510

// In a called workflow, its jobs' keys sit under the job that called it.

2511

let parent = job.call().filter(|c| c["role"] == "callee").and_then(|c| c["parent"].as_str().map(str::to_owned));

2512

for need in &spec.needs {

2513

let key = match &parent {

2514

Some(parent) => format!("{parent}/{need}"),

2515

None => need.clone(),

2516

};

2517

let rows: Vec<&JobRow> = jobs.iter().filter(|row| row.key == key).collect();

2518

let mut outputs = Map::new();

2519

for row in &rows {

2520

if let Ok(Value::Object(more)) = serde_json::from_str::<Value>(&row.outputs) {

2521

outputs.extend(more);

2522

}

2523

}

2524

needs.insert(need.clone(), json!({ "result": key_result(&rows), "outputs": outputs }));

2525

}

2526

let siblings = jobs.iter().filter(|row| row.key == job.key).count();

2527

let matrix: Value = job.matrix.as_deref().and_then(|m| serde_json::from_str(m).ok()).unwrap_or(json!({}));

2528

let info = run.info();

2529

let mut github = info.context(&job.key, &token, run.action.as_deref());

2530

github["token"] = json!(token);

2531

github["retention_days"] = json!(retention_days);

2532

// On a self-hosted runner, `runner` and `RUNNER_*` describe that

2533

// machine rather than g1t's sandbox.

2534

let mut variables = info.variables(&job.key);

2535

variables.insert("GITHUB_RETENTION_DAYS".into(), json!(retention_days.to_string()));

2536

let mut runner = match &job.runner_id {

2537

Some(id) => self.runner_context_for(id, &mut variables).await?,

2538

None => runner_context(),

2539

};

2540

// A re-run with debug logging: what GitHub sets for one.

2541

if run.debug != 0 {

2542

debug_logging(&mut variables, &mut runner);

2543

}

2544

2545

// Where to check out: the repository, which holds a pull request's

2546

// test merge commit; a pull request's fork when the run builds its

2547

// head instead (`run.sha` is the head, `info.sha` what is built).

2548

let clone_url = match run.pull {

2549

Some(number) if run.event.starts_with("pull_request") && run.event != "pull_request_target" && info.sha == run.sha => {

2550

let located: Outcome<g1t_contracts::work::PullDetail> = g1t_kit::call(

2551

&self.work,

2552

"get_pull",

2553

&g1t_contracts::work::ViewArgs {

2554

repo: repo.clone(),

2555

number,

2556

viewer: self.workspace_actor(&repo.namespace).await?,

after_seq: 0,

},

)

.await?;

match located {

Outcome::Ok(detail) => match detail.pull.fork.or(detail.pull.head_repo) {

2563

Some(fork) => format!("{SITE}/{}/{}.git", fork.namespace, fork.name),

2564

None => format!("{SITE}/{}.git", run.repo),

2565

},

2566

Outcome::Fail(_) => format!("{SITE}/{}.git", run.repo),

2567

}

2568

}

2569

_ => format!("{SITE}/{}.git", run.repo),

2570

};

2571

2572

// Only one fetch with a token gets its spec: of two at once, the

2573

// second finds the hash already changed and is refused.

if next.is_some() {

let swapped = self

.db

.prepare(TOKEN_ROTATE)

2578

.bind(&[token_hash.as_str().into(), job.id.as_str().into(), job.token_hash.as_deref().unwrap_or_default().into()])?

2579

.first::<Value>(None)

2580

.await?;

2581

if swapped.is_none() {

2582

return Ok(fail(FailureCode::Unauthenticated, "That job is not running, or the token is not its."));

}

}

let mut spec = json!({

"job": job.id,

"run": run.id,

"key": job.key,

"name": job.name,

"spec": spec.raw,

"workflow": {

"env": workflow.env,

"defaults": workflow.raw.get("defaults").cloned().unwrap_or(Value::Null),

2595

},

2596

"github": github,

2597

"variables": variables,

"event": info.event,

"contexts": {

"vars": vars,

"secrets": secrets,

"inputs": call_inputs.unwrap_or_else(|| Value::Object(run.inputs())),

"matrix": matrix,

"needs": needs,

"strategy": {

"fail-fast": spec.fail_fast,

2607

"job-index": job.ordinal,

2608

"job-total": siblings,

2609

"max-parallel": spec.max_parallel.unwrap_or(siblings as u32),

},

"runner": runner,

},

"checkout": {

"repository": run.repo,

"url": clone_url,

"sha": info.sha,

"ref": run.git_ref,

"token": token,

},

"timeoutMinutes": job.timeout_minutes,

2621

"masks": masks,

2622

// As the job's log lists them at its start.

2623

"permissions": permissions.listed().into_iter().map(|(name, access)| (name.to_owned(), json!(access.as_str()))).collect::<Map<String, Value>>(),

2624

// Whether the job may ask for an OIDC token decides whether it

2625

// is told where to.

2626

"runtime": {

2627

"token": runtime_token,

2628

"idToken": self.oidc_allowed(&run, &job),

2629

},

2630

});

2631

if let Some(next) = next {

2632

spec["nextToken"] = json!(next);

2633

}

2634

if relayed {

2635

spec["brokered"] = broker::spec_entries(&stand_ins);

2636

}

2637

Ok(Outcome::Ok(spec))

2638

}

2639

2640

/// `job_report`: the sandbox telling how the job is going.

2641

pub async fn job_report(&self, a: JobCallArgs) -> Result<Outcome<Value>> {

2642

let job = check!(self.job_for_token(&a).await?);

2643

let report = &a.report;

2644

let at = now();

2645

match report["kind"].as_str().unwrap_or_default() {

2646

"steps" => {

2647

// The list can grow as the job goes (post steps), so steps

2648

// already reported keep where they stand.

2649

let known: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();

2650

let steps: Vec<Value> = report["steps"]

.as_array()

.map(|names| {

names

.iter()

.enumerate()

.map(|(i, name)| match known.get(i) {

2657

Some(step) if step["status"] != "queued" => step.clone(),

2658

_ => json!({ "number": i + 1, "name": expr::to_text(name), "status": "queued", "conclusion": null, "startedAt": null, "finishedAt": null }),

})

.collect()

})

.unwrap_or_default();

2663

self.db

2664

.prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")

2665

.bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?

.run()

.await?;

}

"step" => {

let number = report["number"].as_u64().unwrap_or(0) as usize;

2671

let mut steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();

2672

if let Some(step) = number.checked_sub(1).and_then(|i| steps.get_mut(i)) {

2673

let status = report["status"].as_str().unwrap_or("in_progress");

2674

step["status"] = json!(status);

2675

if status == "in_progress" {

2676

step["startedAt"] = json!(at);

2677

}

2678

if status == "completed" {

2679

step["finishedAt"] = json!(at);

2680

step["conclusion"] = report["conclusion"].clone();

2681

}

2682

if let Some(name) = report["name"].as_str() {

2683

step["name"] = json!(name);

}

}

self.db

.prepare("UPDATE jobs SET steps = ?, seen_at = ? WHERE id = ?")

2688

.bind(&[serde_json::to_string(&steps)?.into(), at.as_str().into(), job.id.as_str().into()])?

.run()

.await?;

}

// A line from g1t itself, not the job: the runner Worker's note

2693

// that a brokered secret went to a host it is not for. Put

2694

// under the step that is running.

2695

"note" => {

2696

let text = report["text"].as_str().unwrap_or_default().chars().take(1000).collect::<String>();

2697

let steps: Vec<Value> = serde_json::from_str(&job.steps).unwrap_or_default();

2698

let step = steps.iter().rev().find(|step| step["status"] == "in_progress").and_then(|step| step["number"].as_u64()).unwrap_or(0);

2699

if !text.trim().is_empty() {

2700

self.note_log(&job.id, step as u32, &text).await?;

}

}

"log" => {

let mut text = report["text"].as_str().unwrap_or_default().to_owned();

2705

// A brokered secret's value never enters the sandbox, but an

2706

// answer that echoed it whole could carry it in; it is

2707

// masked here, where the value is known.

2708

if job.brokered != 0 {

2709

let values: Vec<String> = self.stand_ins_of(&job.id).await?.into_iter().map(|secret| secret.value).collect();

2710

if !values.is_empty() {

2711

text = g1t_actions::mask::apply(&text, &g1t_actions::mask::all_variants(values.iter().map(String::as_str)));

2712

}

2713

}

2714

if text.len() > MAX_CHUNK_BYTES {

2715

let mut cut = MAX_CHUNK_BYTES;

2716

while !text.is_char_boundary(cut) {

cut -= 1;

}

text.truncate(cut);

}

#[derive(Deserialize)]

struct Size {

n: Option<f64>,

}

let size = self

.db

.prepare("SELECT SUM(LENGTH(text)) AS n FROM logs WHERE job_id = ?")

2728

.bind(&[job.id.as_str().into()])?

2729

.first::<Size>(None)

2730

.await?;

2731

let used = size.and_then(|s| s.n).unwrap_or(0.0) as usize;

2732

if used < MAX_LOG_BYTES {

2733

if used + text.len() >= MAX_LOG_BYTES {

2734

text.push_str("\n… The log reached its limit of 4 MB; the rest is not kept.\n");

2735

}

2736

// The next `seq` is taken in the insert itself, so two

2737

// batches at once (a report sent again while the first

2738

// was still being written) never reach for the same one

2739

// and lose a chunk to the primary key.

self.db

.prepare(LOG_INSERT)

.bind(&[

job.id.as_str().into(),

2744

(report["step"].as_u64().unwrap_or(0) as u32).into(),

2745

text.into(),

2746

report["chunk"].as_u64().map_or(worker::wasm_bindgen::JsValue::NULL, |n| (n as f64).into()),

])?

.run()

.await?;

}

self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;

2752

}

2753

"summary" => {

2754

// `$GITHUB_STEP_SUMMARY`, masked by the runner: added to the

2755

// step's summary, up to 1 MiB a step and 20 steps a job.

2756

let step = report["step"].as_u64().unwrap_or(0) as u32;

2757

let markdown = report["markdown"].as_str().unwrap_or_default();

2758

#[derive(Deserialize)]

struct Held {

steps: u32,

mine: Option<f64>,

}

let held = self

.db

.prepare("SELECT COUNT(*) AS steps, MAX(CASE WHEN step = ? THEN LENGTH(markdown) END) AS mine FROM job_summaries WHERE job_id = ?")

2766

.bind(&[step.into(), job.id.as_str().into()])?

2767

.first::<Held>(None)

2768

.await?;

2769

let (steps, mine) = held.map_or((0, None), |held| (held.steps, held.mine.map(|n| n as usize)));

2770

if summary_fits(steps, mine, markdown.len()) {

2771

self.db

2772

.prepare(

2773

"INSERT INTO job_summaries (job_id, step, markdown) VALUES (?, ?, ?)

2774

ON CONFLICT (job_id, step) DO UPDATE SET markdown = job_summaries.markdown || excluded.markdown",

2775

)

2776

.bind(&[job.id.as_str().into(), step.into(), markdown.into()])?

.run()

.await?;

}

self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;

2781

}

2782

"annotation" => {

2783

let mut annotations: Vec<Value> = serde_json::from_str(&job.annotations).unwrap_or_default();

2784

if annotations.len() < MAX_ANNOTATIONS {

2785

annotations.push(json!({

2786

"level": report["level"].as_str().unwrap_or("notice"),

2787

"message": report["message"].as_str().unwrap_or_default().chars().take(4000).collect::<String>(),

2788

"title": report["title"],

2789

"file": report["file"],

2790

"line": report["line"],

2791

}));

2792

self.db

2793

.prepare("UPDATE jobs SET annotations = ?, seen_at = ? WHERE id = ?")

2794

.bind(&[serde_json::to_string(&annotations)?.into(), at.as_str().into(), job.id.as_str().into()])?

.run()

.await?;

}

}

// Nothing to tell; the answer says whether to stop.

2800

"ping" => {

2801

self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[at.into(), job.id.as_str().into()])?.run().await?;

2802

}

2803

"done" => {

2804

// A job told to stop ends cancelled, however its cleanup went.

2805

let conclusion = if job.cancel_requested_at.is_some() {

"cancelled"

} else {

report["conclusion"]

.as_str()

.filter(|c| matches!(*c, "success" | "failure" | "cancelled"))

2811

.unwrap_or("failure")

2812

};

2813

let outputs = report["outputs"].as_object().cloned();

2814

Box::pin(self.finish_job(&job.id, conclusion, report["reason"].as_str(), outputs.as_ref())).await?;

2815

}

2816

other => return Ok(fail(FailureCode::Invalid, format!("There is no report called `{other}`."))),

2817

}

2818

// `cancelled`: the run was cancelled, and the runner should stop

2819

// the step it is on and run only its cleanup steps.

2820

Ok(Outcome::Ok(json!({ "ok": true, "cancelled": job.cancel_requested_at.is_some() })))

2821

}

2822

2823

// --- Every minute ---------------------------------------------------------------

2824

2825

pub async fn on_minute(&self, now_ms: u64) -> Result<()> {

2826

let minute = now_ms / 60_000 * 60_000;

2827

// Repositories a workspace's managed workflows changed in, beyond

2828

// those synced at once (managed.rs): their schedules wait for it.

2829

if let Err(error) = self.catch_up_syncs().await {

2830

worker::console_error!("actions: catching up syncs failed: {error}");

2831

}

2832

// Staff (or billing's usage watcher) paused scheduled runs across

2833

// g1t: this minute's schedules are skipped, not queued for later.

2834

// Kept 30 seconds in the isolate (g1t_kit::pause).

2835

if g1t_kit::pause::paused(&self.billing, g1t_contracts::billing::PauseLevel::Schedules).await {

2836

worker::console_log!("actions: schedules are paused across g1t; skipped this minute's");

2837

} else if let Err(error) = self.run_schedules(minute).await {

2838

worker::console_error!("actions: schedules failed: {error}");

2839

}

2840

// Jobs whose sandbox went quiet or ran past their time.

2841

let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;

2842

for job in running {

2843

// Times in g1t's format compare as text.

2844

let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));

2845

let silent = job.seen_at.as_deref().is_some_and(|seen| seen < before(SILENT_MS).as_str());

2846

let limit = (u64::from(job.timeout_minutes) * 60 + 120) * 1000;

2847

let over = job.started_at.as_deref().is_some_and(|started| started < before(limit).as_str());

2848

let cancel_overdue = job.cancel_requested_at.as_deref().is_some_and(|at| at < before(CANCEL_GRACE_MS).as_str());

2849

if cancel_overdue {

2850

// Cancelled, and still going after its grace period.

2851

self.hard_stop(&job, "It was cancelled, and did not finish its cleanup steps within 5 minutes.").await?;

2852

self.advance(&job.run_id).await?;

2853

} else if over {

2854

let reason = format!("It ran longer than its time limit of {} minutes.", job.timeout_minutes);

2855

// A self-hosted runner is told to stop on its next poll.

2856

if job.runner_id.is_none() {

2857

let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;

2858

}

2859

self.finish_job(&job.id, "failure", Some(&reason), None).await?;

2860

} else if silent {

2861

let reason = match &job.runner_name {

2862

Some(name) => format!("The self-hosted runner {name} stopped answering."),

2863

None => "The runner stopped answering.".to_owned(),

2864

};

2865

self.finish_job(&job.id, "failure", Some(&reason), None).await?;

2866

}

2867

}

2868

// Calls of reusable workflows waiting in a concurrency group, in

2869

// case the job they waited for was cancelled in another run.

2870

#[derive(Deserialize)]

struct RunId {

run_id: String,

}

let held = self

.db

.prepare("SELECT DISTINCT run_id FROM jobs WHERE status = 'waiting' AND concurrency_group IS NOT NULL LIMIT 20")

2877

.all()

2878

.await?

2879

.results::<RunId>()?;

2880

for run in held {

2881

self.advance(&run.run_id).await?;

2882

}

2883

if let Err(error) = self.sweep_runners(now_ms).await {

2884

worker::console_error!("actions: the runners' sweep failed: {error}");

2885

}

2886

// Jobs held at an environment whose wait timer has run out.

2887

if let Err(error) = self.release_gates().await {

2888

worker::console_error!("actions: environments' gates failed: {error}");

2889

}

2890

// Once an hour: the cache's expired entries, and its storage.

2891

if (now_ms / 60_000) % 60 == 7

2892

&& let Err(error) = self.sweep_cache(now_ms).await

2893

{

2894

worker::console_error!("actions: the cache's sweep failed: {error}");

2895

}

2896

// And the archives of GitHub's actions no job uses any more.

2897

if (now_ms / 60_000) % 60 == 47

2898

&& let Err(error) = self.sweep_action_archives(now_ms).await

2899

{

2900

worker::console_error!("actions: the action archives' sweep failed: {error}");

2901

}

2902

// And artifacts past their time, and the toolkit's abandoned parts.

2903

if (now_ms / 60_000) % 60 == 37 {

2904

if let Err(error) = self.sweep_artifacts(now_ms).await {

2905

worker::console_error!("actions: the artifacts' sweep failed: {error}");

2906

}

2907

if let Err(error) = self.sweep_blob_parts(now_ms).await {

2908

worker::console_error!("actions: the blob parts' sweep failed: {error}");

2909

}

2910

}

2911

// And logs of jobs that finished longer ago than LOG_RETENTION_DAYS.

2912

if (now_ms / 60_000) % 60 == 17

2913

&& let Err(error) = self.sweep_logs(now_ms).await

2914

{

2915

worker::console_error!("actions: the logs' sweep failed: {error}");

2916

}

2917

self.start_queued().await

}

}

#[cfg(test)]

mod log_chunks {

use rusqlite::{Connection, params};

2925

2926

use super::LOG_INSERT;

2927

2928

/// The logs table as the migrations leave it.

2929

fn database() -> Connection {

2930

let db = Connection::open_in_memory().unwrap();

2931

for name in ["0001_init.sql", "0016_log_chunks.sql"] {

2932

let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));

2933

db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();

}

db

}

fn log(db: &Connection, job: &str) -> Vec<(i64, i64, String)> {

2939

let mut rows = db.prepare("SELECT seq, step, text FROM logs WHERE job_id = ? ORDER BY seq").unwrap();

2940

rows.query_map([job], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).unwrap().map(Result::unwrap).collect()

}

#[test]

fn a_batch_sent_again_is_kept_once() {

2945

let db = database();

2946

let insert = |job: &str, step: i64, text: &str, chunk: Option<i64>| db.execute(LOG_INSERT, params![job, step, text, chunk]).unwrap();

2947

assert_eq!(insert("job_1", 0, "one\n", Some(1)), 1);

2948

// The runner did not hear the answer and sent the batch again.

2949

assert_eq!(insert("job_1", 0, "one\n", Some(1)), 0);

2950

assert_eq!(insert("job_1", 1, "two\n", Some(2)), 1);

2951

// Another job numbers its batches from 1 too.

2952

assert_eq!(insert("job_2", 0, "other\n", Some(1)), 1);

2953

// A runner that numbers nothing has every batch kept, as before.

2954

assert_eq!(insert("job_3", 0, "a\n", None), 1);

2955

assert_eq!(insert("job_3", 0, "a\n", None), 1);

2956

assert_eq!(log(&db, "job_1"), [(1, 0, "one\n".into()), (2, 1, "two\n".into())]);

2957

assert_eq!(log(&db, "job_2"), [(1, 0, "other\n".into())]);

2958

assert_eq!(log(&db, "job_3").len(), 2);

2959

// A job run again moves its log aside (`{id}.{attempt}`), and the

2960

// new attempt's runner numbers from 1 again.

2961

db.execute("UPDATE logs SET job_id = job_id || '.1' WHERE job_id = 'job_1'", []).unwrap();

2962

assert_eq!(insert("job_1", 0, "again\n", Some(1)), 1);

2963

assert_eq!(log(&db, "job_1"), [(1, 0, "again\n".into())]);

2964

assert_eq!(log(&db, "job_1.1").len(), 2);

}

}

#[cfg(test)]

mod token_rotation {

use g1t_secrets::sha256_hex;

2971

use rusqlite::{Connection, OptionalExtension, params};

2972

2973

use super::{TOKEN_ROTATE, token_opens};

2974

2975

fn database() -> Connection {

2976

let db = Connection::open_in_memory().unwrap();

2977

for name in ["0001_init.sql", "0020_job_token_rotation.sql"] {

2978

let path = format!("{}/migrations/{name}", env!("CARGO_MANIFEST_DIR"));

2979

db.execute_batch(&std::fs::read_to_string(path).unwrap()).unwrap();

2980

}

2981

db.execute(

2982

"INSERT INTO jobs (id, run_id, repo_id, namespace, key, name, status, token_hash) VALUES ('job_1', 'run_1', 'repo_1', 'acme', 'build', 'build', 'in_progress', ?1)",

2983

[sha256_hex("first")],

)

.unwrap();

db

}

fn rotate(db: &Connection, next: &str, shown: &str) -> bool {

2990

db.query_row(TOKEN_ROTATE, params![sha256_hex(next), "job_1", sha256_hex(shown)], |row| row.get::<_, String>(0))

.optional()

.unwrap()

.is_some()

}

/// The job's current token's hash, and the one it started with.

2997

fn hashes(db: &Connection) -> (Option<String>, Option<String>) {

2998

db.query_row("SELECT token_hash, first_token_hash FROM jobs WHERE id = 'job_1'", [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()

}

#[test]

fn a_job_token_fetches_its_spec_once() {

3003

let db = database();

3004

assert!(rotate(&db, "second", "first"));

3005

assert_eq!(hashes(&db), (Some(sha256_hex("second")), Some(sha256_hex("first"))));

3006

// The first token, shown again (a copy left on the computer, or a

3007

// second fetch racing the first), swaps nothing and is refused.

3008

assert!(!rotate(&db, "third", "first"));

3009

assert_eq!(hashes(&db).0, Some(sha256_hex("second")));

3010

// A job that ended has no token to swap.

3011

db.execute("UPDATE jobs SET status = 'completed', token_hash = NULL, first_token_hash = NULL", []).unwrap();

3012

assert!(!rotate(&db, "fourth", "second"));

3013

assert_eq!(hashes(&db), (None, None));

}

#[test]

fn the_worker_finds_brokered_secrets_with_the_token_the_job_started_with() {

3018

let db = database();

3019

let opens = |db: &Connection, token: &str, started: bool| {

3020

let (current, first) = hashes(db);

3021

token_opens(current.as_deref(), first.as_deref(), token, started)

3022

};

3023

// Before any swap, the job's one token opens everything.

3024

assert!(opens(&db, "first", false) && opens(&db, "first", true));

3025

assert!(rotate(&db, "second", "first"));

3026

// The harness's new token opens everything.

3027

assert!(opens(&db, "second", false) && opens(&db, "second", true));

3028

// The token it started with, which the runner Worker holds, still

3029

// finds the job's brokered secrets, and opens nothing else.

3030

assert!(opens(&db, "first", true));

3031

assert!(!opens(&db, "first", false));

3032

assert!(!opens(&db, "other", true));

3033

// Another swap keeps the token it started with, not the one between.

3034

assert!(rotate(&db, "third", "second"));

3035

assert!(opens(&db, "first", true));

3036

assert!(!opens(&db, "second", true));

}

#[test]

fn the_runtime_token_is_signed_for_the_token_that_replaces_the_first() {

3041

let now = 1_800_000_000;

3042

let token = crate::runtime::runtime_token("job_1", "run_1", &sha256_hex("second"), now, 600);

3043

assert!(crate::runtime::runtime_valid(&token, "job_1", &sha256_hex("second"), now));

3044

assert!(!crate::runtime::runtime_valid(&token, "job_1", &sha256_hex("first"), now));

}

}

#[cfg(test)]

mod stopping {

use super::stops_runs;

3051

use g1t_contracts::events::Event;

3052

use serde_json::{Value, json};

3053

3054

fn event(kind: &str, data: Value) -> Event {

Event {

id: "evt_1".into(),

kind: kind.into(),

source: "repos".into(),

3059

time: "2026-10-05T00:00:00Z".into(),

3060

repo_id: Some("rep_1".into()),

actor: None,

data,

}

}

#[test]

fn deleting_or_archiving_stops_runs() {

3068

assert_eq!(stops_runs(&event("repo.deleted", json!({ "repoId": "rep_1" }))).as_deref(), Some("rep_1"));

3069

assert_eq!(stops_runs(&event("repo.archived", json!({ "archived": true }))).as_deref(), Some("rep_1"));

3070

assert_eq!(stops_runs(&event("repo.unarchived", json!({ "archived": false }))), None);

3071

assert_eq!(stops_runs(&event("repo.restored", json!({}))), None);

3072

assert_eq!(stops_runs(&event("git.push", json!({}))), None);

}

}

#[cfg(test)]

mod status_of_needs {

3078

use std::collections::HashMap;

3079

3080

use super::ancestor_failed;

3081

3082

/// check -> plan -> (migrate) -> core -> edge, as deploy.yml has them,

3083

/// and a job that needs only the last.

3084

fn graph() -> HashMap<&'static str, Vec<&'static str>> {

3085

HashMap::from([

3086

("check", vec![]),

3087

("plan", vec!["check"]),

3088

("migrate", vec!["plan"]),

3089

("core", vec!["plan", "migrate"]),

3090

("edge", vec!["plan", "migrate", "core"]),

3091

("notify", vec!["edge"]),

])

}

#[test]

fn a_failure_is_seen_however_far_back() {

3097

let needs = graph();

3098

let failed = |which: &'static str| move |key: &str| key == which;

3099

// check failed; plan, and everything after, was skipped for it.

3100

assert!(ancestor_failed(&needs, "notify", failed("check")));

3101

assert!(ancestor_failed(&needs, "core", failed("check")));

3102

assert!(ancestor_failed(&needs, "edge", failed("core")));

3103

// Nothing before a job failed: a skipped migrate is not a failure.

3104

assert!(!ancestor_failed(&needs, "edge", |_| false));

3105

assert!(!ancestor_failed(&needs, "core", failed("edge")));

3106

assert!(!ancestor_failed(&needs, "check", failed("check")));

}

#[test]

fn cycles_and_unknown_keys_end() {

3111

let needs = HashMap::from([("a", vec!["b"]), ("b", vec!["a"])]);

3112

assert!(!ancestor_failed(&needs, "a", |_| false));

3113

assert!(!ancestor_failed(&needs, "missing", |_| true));

}

}

#[cfg(test)]

mod deployments {

use serde_json::{Map, Value, json};

3120

3121

use super::{JobEnvironment, deployment_outcome, environment_of};

3122

3123

#[test]

3124

fn a_jobs_environment_is_read_for_deployments() {

3125

let contexts: Map<String, Value> = serde_json::from_value(json!({

3126

"github": { "ref_name": "main", "repository": "acme/web" },

3127

"inputs": { "target": "staging" },

"matrix": {},

}))

.unwrap();

let read = |raw: Value| environment_of(&raw, &contexts);

3132

assert_eq!(read(json!({})), None);

3133

assert_eq!(

3134

read(json!({ "environment": "production" })),

3135

Some(JobEnvironment { name: "production".into(), url: None, deploys: true })

3136

);

3137

assert_eq!(

3138

read(json!({ "environment": { "name": "production", "url": "https://g1t.sh" } })),

3139

Some(JobEnvironment { name: "production".into(), url: Some("https://g1t.sh".into()), deploys: true })

3140

);

3141

// Expressions are filled in from the run.

3142

assert_eq!(

3143

read(json!({ "environment": { "name": "${{ inputs.target }}", "url": "https://${{ github.ref_name }}.example.com" } })),

3144

Some(JobEnvironment { name: "staging".into(), url: Some("https://main.example.com".into()), deploys: true })

3145

);

3146

// Secrets only: no deployment.

3147

assert!(!read(json!({ "environment": { "name": "production", "deployment": false } })).unwrap().deploys);

3148

// The same, said in the job's g1t settings.

3149

assert!(!read(json!({ "environment": "production", "g1t": { "deployment": false } })).unwrap().deploys);

3150

assert!(!read(json!({ "environment": { "name": "production" }, "g1t": { "deployment": false } })).unwrap().deploys);

3151

// Only http(s) addresses.

3152

assert_eq!(read(json!({ "environment": { "name": "production", "url": "javascript:alert(1)" } })).unwrap().url, None);

}

#[test]

fn a_runs_outcome_for_an_environment() {

3157

let of = |list: &[&str]| deployment_outcome(&list.iter().map(|c| Some((*c).to_owned())).collect::<Vec<_>>());

3158

assert_eq!(of(&["success", "skipped"]), Some("success"));

3159

assert_eq!(of(&["success", "failure"]), Some("failure"));

3160

assert_eq!(of(&["success", "cancelled"]), Some("error"));

3161

assert_eq!(of(&["skipped"]), None);

3162

assert_eq!(deployment_outcome(&[None]), None);

}

}

#[cfg(test)]

mod reruns {

use serde_json::{Map, Value, json};

3169

3170

use super::{MAX_SUMMARIES, MAX_SUMMARY_BYTES, Rerun, Scope, Status, call_allowed, call_prefix, debug_logging, rerun_keys, shown_name, summary_fits, top_key};

3171

3172

/// build ← test ← deploy, and lint on its own.

3173

fn keys(which: Rerun, failed: &[&str]) -> Vec<String> {

3174

let order = ["build", "lint", "test", "deploy"];

3175

let needs = |key: &str| -> Vec<String> {

3176

match key {

3177

"test" => vec!["build".into()],

3178

"deploy" => vec!["test".into()],

_ => Vec::new(),

}

};

rerun_keys(&order, needs, |key| !failed.contains(&key), which)

}

#[test]

fn a_rerun_takes_the_jobs_it_names_and_those_that_need_them() {

3187

assert_eq!(keys(Rerun::All, &[]), ["build", "lint", "test", "deploy"]);

3188

assert_eq!(keys(Rerun::Failed, &["test"]), ["test", "deploy"]);

3189

assert_eq!(keys(Rerun::Failed, &["lint"]), ["lint"]);

3190

assert!(keys(Rerun::Failed, &[]).is_empty());

3191

// One job, whatever it came to, and what depends on it.

3192

assert_eq!(keys(Rerun::Job("build"), &[]), ["build", "test", "deploy"]);

3193

assert_eq!(keys(Rerun::Job("lint"), &["test"]), ["lint"]);

3194

assert_eq!(keys(Rerun::Job("deploy"), &[]), ["deploy"]);

3195

assert!(keys(Rerun::Job("missing"), &[]).is_empty());

}

#[test]

fn a_called_workflows_jobs_run_again_with_the_job_that_calls_it() {

3200

assert_eq!(top_key("build/test"), "build");

3201

assert_eq!(top_key("build/inner/test"), "build");

3202

assert_eq!(top_key("lint"), "lint");

3203

// A matrix's calls each have their own jobs, under `key[i]/`.

3204

assert_eq!(top_key("build[1]/test"), "build");

3205

assert_eq!(top_key("build/inner[0]/test"), "build");

3206

assert_eq!(call_prefix("build", Some(1)), "build[1]");

3207

assert_eq!(call_prefix("build/inner", None), "build/inner");

}

#[test]

fn a_run_connects_ten_levels_of_workflows_and_fifty_reusable_ones() {

3212

assert!(call_allowed(1, &[], "a.yml").is_ok());

3213

assert!(call_allowed(9, &[], "a.yml").is_ok());

3214

let deep = call_allowed(10, &[], "a.yml").unwrap_err();

3215

assert!(deep.contains("at most ten") && deep.contains("nine levels"), "{deep}");

3216

let names: Vec<String> = (0..50).map(|n| format!("w{n}.yml")).collect();

3217

let fifty: Vec<&str> = names.iter().map(String::as_str).collect();

3218

// A workflow called again, as by each combination of a matrix, is one.

3219

assert!(call_allowed(1, &fifty, "w3.yml").is_ok());

3220

assert!(call_allowed(1, &fifty, "another.yml").unwrap_err().contains("at most 50 different"));

}

#[test]

fn a_matrixs_calls_are_named_by_the_caller_and_its_values() {

3225

let workflow = super::workflow::parse(

3226

"on: push\njobs:\n build:\n name: Build\n strategy:\n matrix: { os: [linux], node: [20] }\n uses: ./.g1t/workflows/b.yml\n plain:\n uses: ./.g1t/workflows/b.yml\n test:\n name: Test ${{ matrix.os }}\n strategy:\n matrix: { os: [linux] }\n runs-on: ubuntu-latest\n steps: [{ run: 'true' }]",

3227

)

3228

.unwrap();

3229

let job = |id: &str| workflow.jobs.iter().find(|job| job.id == id).unwrap();

3230

let combination: Map<String, Value> = serde_json::from_value(json!({ "os": "linux", "node": 20 })).unwrap();

3231

let mut contexts = Map::new();

3232

contexts.insert("matrix".into(), Value::Object(combination.clone()));

3233

let scope = Scope { contexts: &contexts, status: Status::Success, hash_files: None };

3234

assert_eq!(shown_name(job("build"), None, &combination, &scope), "Build (linux, 20)");

3235

assert_eq!(shown_name(job("plain"), None, &Map::new(), &scope), "plain");

3236

// A called workflow's job, under the call that made it.

3237

assert_eq!(shown_name(job("plain"), Some("Build (linux, 20)"), &Map::new(), &scope), "Build (linux, 20) / plain");

3238

assert_eq!(shown_name(job("test"), Some("Build"), &combination, &scope), "Build / Test linux");

}

#[test]

fn a_job_keeps_twenty_steps_summaries_of_a_mebibyte_each() {

3243

assert!(summary_fits(0, None, 10));

3244

assert!(!summary_fits(0, None, 0), "nothing to keep");

3245

assert!(!summary_fits(MAX_SUMMARIES, None, 10), "a twenty-first step's summary is dropped");

3246

assert!(summary_fits(MAX_SUMMARIES, Some(10), 10), "a step that has one may add to it");

3247

assert!(summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 10));

3248

assert!(!summary_fits(1, Some(MAX_SUMMARY_BYTES - 10), 11));

3249

assert!(!summary_fits(0, None, MAX_SUMMARY_BYTES + 1));

}

#[test]

fn a_debug_rerun_sets_what_github_sets() {

3254

let mut variables = Map::new();

3255

let mut runner = json!({ "name": "g1t", "debug": "" });

3256

debug_logging(&mut variables, &mut runner);

3257

assert_eq!(variables["RUNNER_DEBUG"], "1");

3258

assert_eq!(variables["ACTIONS_STEP_DEBUG"], "true");

3259

assert_eq!(variables["ACTIONS_RUNNER_DEBUG"], "true");

3260

assert_eq!(runner["debug"], Value::String("1".into()));

3261

assert_eq!(runner["name"], "g1t");

}

}

#[cfg(test)]

mod machines {

use serde_json::json;

3268

3269

use super::{RunInfo, RunRow, run_is_public};

3270

3271

fn run(event: serde_json::Value) -> RunRow {

3272

let info = serde_json::to_string(&RunInfo { repository: "acme/web".into(), event, ..RunInfo::default() }).unwrap();

3273

serde_json::from_value(json!({

3274

"id": "run_1", "workflow_id": "wf_1", "repo_id": "rep_1", "repo": "acme/web", "path": ".g1t/workflows/ci.yml", "name": "CI",

3275

"title": "CI", "number": 1, "attempt": 1, "event": "push", "git_ref": "refs/heads/main", "sha": "abc", "status": "queued",

3276

"source": "", "info": info, "inputs": "{}", "trusted": 1, "created_at": "2026-10-18T00:00:00Z",

}))

.unwrap()

}

#[test]

fn a_runs_repository_is_public_only_when_its_event_says_so() {

3283

assert!(run_is_public(&run(json!({ "repository": { "private": false } }))));

3284

assert!(!run_is_public(&run(json!({ "repository": { "private": true } }))));

3285

// An event that doesn't say is taken as private: the standard machine.

3286

assert!(!run_is_public(&run(json!({}))));

3287

}

3288

}