9301161
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Mutex, OnceLock};
use ci_protocol::runner::{JobPayload, StepPayload};
use config::runner::RunnerConfig;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tracing::Instrument as _;
pub const MAX_STEP_OUTPUT_BYTES: usize = 4 * 1024 * 1024;
#[derive(Debug)]
pub enum JobOutcome {
Success {
exit_code: Option<i32>,
},
Failure {
exit_code: Option<i32>,
error: Option<String>,
},
Cancelled,
TimedOut,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StepSignal {
Started,
Finished,
}
static CANCELLED: OnceLock<Mutex<HashMap<uuid::Uuid, ()>>> = OnceLock::new();
fn cancelled_set() -> &'static Mutex<HashMap<uuid::Uuid, ()>> {
CANCELLED.get_or_init(|| Mutex::new(HashMap::new()))
}
pub fn request_cancel(job_ids: &[uuid::Uuid]) {
if let Ok(mut set) = cancelled_set().lock() {
for id in job_ids {
set.insert(*id, ());
}
}
#[cfg(unix)]
{
if let Ok(pids) = running_pids().lock()
&& let Some(pid) = job_ids.iter().find_map(|id| pids.get(id).copied())
{
let _ = nix_kill_group(pid);
}
}
}
pub fn clear_cancel(job_id: &uuid::Uuid) {
if let Ok(mut set) = cancelled_set().lock() {
set.remove(job_id);
}
}
pub(crate) fn is_cancelled(job_id: &uuid::Uuid) -> bool {
cancelled_set()
.lock()
.map(|set| set.contains_key(job_id))
.unwrap_or(false)
}
static RUNNING_PIDS: OnceLock<Mutex<HashMap<uuid::Uuid, i32>>> = OnceLock::new();
fn running_pids() -> &'static Mutex<HashMap<uuid::Uuid, i32>> {
RUNNING_PIDS.get_or_init(|| Mutex::new(HashMap::new()))
}
#[cfg(unix)]
fn nix_kill_group(pid: i32) {
// TERM the whole process group so shell children die with the step.
unsafe {
libc_kill(-pid, 15);
}
}
#[cfg(unix)]
unsafe fn libc_kill(pid: i32, sig: i32) {
unsafe extern "C" {
fn kill(pid: i32, sig: i32) -> i32;
}
unsafe {
kill(pid, sig);
}
}
#[derive(Clone)]
pub struct StepHooks {
pub on_status: std::sync::Arc<dyn Fn(i32, StepSignal, Option<i32>) + Send + Sync>,
pub on_log: std::sync::Arc<dyn Fn(String) + Send + Sync>,
}
#[allow(clippy::too_many_arguments)]
pub async fn execute_job(
cfg: &RunnerConfig,
payload: &JobPayload,
hooks: StepHooks,
on_artifact: std::sync::Arc<dyn Fn(String, String, i64, Vec<u8>) + Send + Sync>,
renew: std::sync::Arc<dyn Fn() + Send + Sync>,
) -> JobOutcome {
let workspace = PathBuf::from(&cfg.workspace_root)
.join(payload.run_id.to_string())
.join(payload.job_id.to_string());
if let Err(error) = std::fs::create_dir_all(&workspace) {
return JobOutcome::Failure {
exit_code: None,
error: Some(format!("creating workspace failed: {error}")),
};
}
let repo_dir = workspace.join("repo");
if let Err(error) = prepare_workspace(cfg, payload, &workspace, &repo_dir, hooks.clone()).await
{
cleanup(cfg, &workspace);
return JobOutcome::Failure {
exit_code: None,
error: Some(error.to_string()),
};
}
if let Err(error) = download_needs_artifacts(payload, &repo_dir, hooks.clone()).await {
(hooks.on_log)(format!("artifact download failed: {error}\n"));
}
let mut exit_code = 0i32;
let mut failure: Option<String> = None;
let deadline =
std::time::Instant::now() + std::time::Duration::from_secs(payload.timeout_secs.max(1));
for step in &payload.steps {
if is_cancelled(&payload.job_id) {
(hooks.on_status)(step.index, StepSignal::Finished, None);
cleanup(cfg, &workspace);
return JobOutcome::Cancelled;
}
if std::time::Instant::now() >= deadline {
cleanup(cfg, &workspace);
return JobOutcome::TimedOut;
}
(hooks.on_status)(step.index, StepSignal::Started, None);
(hooks.on_log)(format!(
"::group:: step {} · {}\n",
step.index + 1,
step.name
));
let step_span = tracing::info_span!("ci.step", index = step.index, name = %step.name);
let result = run_step(cfg, payload, step, &repo_dir, hooks.clone(), deadline)
.instrument(step_span)
.await;
match result {
Ok(code) => {
(hooks.on_status)(step.index, StepSignal::Finished, Some(code));
(hooks.on_log)("::endgroup::\n".to_string());
if code != 0 {
exit_code = code;
failure = Some(format!("step `{}` exited with {code}", step.name));
break;
}
}
Err(StepFailure::Cancelled) => {
(hooks.on_status)(step.index, StepSignal::Finished, None);
cleanup(cfg, &workspace);
return JobOutcome::Cancelled;
}
Err(StepFailure::TimedOut) => {
(hooks.on_status)(step.index, StepSignal::Finished, None);
cleanup(cfg, &workspace);
return JobOutcome::TimedOut;
}
Err(StepFailure::Io(message)) => {
(hooks.on_status)(step.index, StepSignal::Finished, None);
(hooks.on_log)(format!("{message}\n"));
exit_code = 1;
failure = Some(message);
break;
}
}
(renew)();
}
collect_artifacts(cfg, payload, &repo_dir, Arc::clone(&on_artifact)).await;
cleanup(cfg, &workspace);
match failure {
None => JobOutcome::Success {
exit_code: Some(exit_code),
},
Some(error) => JobOutcome::Failure {
exit_code: Some(exit_code),
error: Some(error),
},
}
}
enum StepFailure {
Cancelled,
TimedOut,
Io(String),
}
async fn prepare_workspace(
cfg: &RunnerConfig,
payload: &JobPayload,
workspace: &Path,
repo_dir: &Path,
hooks: StepHooks,
) -> anyhow::Result<()> {
(hooks.on_log)(format!(
"clone {} @ {}\n",
payload.clone_url, payload.commit_sha
));
if repo_dir.exists() {
std::fs::remove_dir_all(repo_dir)?;
}
let auth_url = match &cfg.pat {
pat if !pat.trim().is_empty() => inject_pat(&payload.clone_url, pat),
_ => payload.clone_url.clone(),
};
let mut command = Command::new(&cfg.git_binary);
command
.arg("clone")
.arg("--no-tags")
.arg(&auth_url)
.arg(repo_dir)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.current_dir(workspace);
run_streaming(command, None, &cfg.clone_timeout_secs, hooks.clone()).await?;
let mut checkout = Command::new(&cfg.git_binary);
checkout
.arg("-C")
.arg(repo_dir)
.arg("checkout")
.arg("--force")
.arg(&payload.commit_sha)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
run_streaming(checkout, None, &cfg.clone_timeout_secs, hooks.clone()).await?;
Ok(())
}
fn inject_pat(url: &str, pat: &str) -> String {
match url.split_once("://") {
Some((scheme, rest)) => format!("{scheme}://{pat}@{rest}"),
None => url.to_string(),
}
}
async fn download_needs_artifacts(
payload: &JobPayload,
repo_dir: &Path,
hooks: StepHooks,
) -> anyhow::Result<()> {
for artifact in &payload.needs_artifacts {
let response = reqwest::get(&artifact.download_url).await?;
if !response.status().is_success() {
anyhow::bail!("artifact {} returned {}", artifact.name, response.status());
}
let bytes = response.bytes().await?;
if let Some(expected) = &artifact.sha256 {
let actual = sha256_hex(&bytes);
if &actual != expected {
anyhow::bail!("artifact {} checksum mismatch", artifact.name);
}
}
let dir = repo_dir.join(".gitdata-artifacts");
std::fs::create_dir_all(&dir)?;
std::fs::write(dir.join(&artifact.name), &bytes)?;
(hooks.on_log)(format!(
"downloaded artifact {} ({} bytes) into .gitdata-artifacts/\n",
artifact.name,
bytes.len()
));
}
Ok(())
}
async fn run_step(
cfg: &RunnerConfig,
payload: &JobPayload,
step: &StepPayload,
repo_dir: &Path,
hooks: StepHooks,
job_deadline: std::time::Instant,
) -> Result<i32, StepFailure> {
let workdir = match &step.working_directory {
Some(dir) if !dir.trim().is_empty() => repo_dir.join(dir.trim()),
_ => repo_dir.to_path_buf(),
};
if !workdir.starts_with(repo_dir) {
return Err(StepFailure::Io(
"working-directory escapes the workspace".to_string(),
));
}
let shell = step
.shell
.clone()
.unwrap_or_else(|| cfg.default_shell.clone());
let mut env: HashMap<String, String> = payload.env.clone().into_iter().collect();
env.extend(step.env.clone());
// Secrets ride on the job payload and are layered under the workflow/step
// env so a stored secret cannot be shadowed by a plaintext variable.
for (key, value) in &payload.secrets {
env.insert(key.clone(), value.clone());
}
if payload.allow_git_token && !cfg.pat.trim().is_empty() {
env.insert("GITDATA_TOKEN".to_string(), cfg.pat.clone());
}
let timeout = step
.timeout_secs
.map(|secs| std::time::Duration::from_secs(secs.max(1)))
.unwrap_or_else(|| {
job_deadline
.saturating_duration_since(std::time::Instant::now())
.max(std::time::Duration::from_secs(1))
});
let mut command = if let Some(image) = &payload.container {
if !cfg.docker_enabled {
return Err(StepFailure::Io(
"job requires a container but docker is disabled on this runner".to_string(),
));
}
let mut cmd = Command::new(&cfg.docker_binary);
cmd.arg("run")
.arg("--rm")
.arg("-v")
.arg(format!("{}:/workspace", repo_dir.display()))
.arg("-w")
.arg(match &step.working_directory {
Some(dir) if !dir.trim().is_empty() => {
format!("/workspace/{}", dir.trim().trim_start_matches('/'))
}
_ => "/workspace".to_string(),
});
for (key, value) in &env {
cmd.arg("-e").arg(format!("{key}={value}"));
}
cmd.arg(image).arg(&shell).arg("-c").arg(&step.run);
cmd
} else {
let mut cmd = Command::new(&shell);
cmd.arg("-c").arg(&step.run).current_dir(&workdir);
for (key, value) in &env {
cmd.env(key, value);
}
cmd
};
command
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.stdin(Stdio::null());
#[cfg(unix)]
{
command.process_group(0);
}
let mut child = command
.spawn()
.map_err(|e| StepFailure::Io(format!("spawning step failed: {e}")))?;
let pid = child.id().map(|p| p as i32);
if let Some(pid) = pid
&& let Ok(mut pids) = running_pids().lock()
{
pids.insert(payload.job_id, pid);
}
let stdout = child.stdout.take();
let stderr = child.stderr.take();
let stdout_hook = Arc::clone(&hooks.on_log);
let stdout_task = stdout.map(|stdout| {
tokio::spawn(async move {
let mut lines = BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
stdout_hook(format!("{line}\n"));
}
})
});
let stderr_hook = Arc::clone(&hooks.on_log);
let stderr_task = stderr.map(|stderr| {
tokio::spawn(async move {
let mut lines = BufReader::new(stderr).lines();
while let Ok(Some(line)) = lines.next_line().await {
stderr_hook(format!("{line}\n"));
}
})
});
// Race the wait against a cancel poll so `docker run` / host steps stop
// promptly on every platform, not only where SIGTERM reaches a group.
let wait: Result<Result<std::process::ExitStatus, std::io::Error>, StepFailure> = tokio::select! {
result = tokio::time::timeout(timeout, child.wait()) => result.map_err(|_| StepFailure::TimedOut),
_ = async {
while !is_cancelled(&payload.job_id) {
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
}
} => {
let _ = child.kill().await;
let _ = child.wait().await;
Err(StepFailure::Cancelled)
}
};
if let Some(task) = stdout_task {
let _ = task.await;
}
if let Some(task) = stderr_task {
let _ = task.await;
}
if let Some(_pid) = pid
&& let Ok(mut pids) = running_pids().lock()
{
pids.remove(&payload.job_id);
}
match wait {
Ok(Ok(status)) => {
if is_cancelled(&payload.job_id) {
return Err(StepFailure::Cancelled);
}
Ok(status.code().unwrap_or(-1))
}
Ok(Err(error)) => Err(StepFailure::Io(format!("step wait failed: {error}"))),
Err(StepFailure::Cancelled) => {
let _ = child.kill().await;
Err(StepFailure::Cancelled)
}
Err(StepFailure::TimedOut) => {
let _ = child.kill().await;
if is_cancelled(&payload.job_id) {
Err(StepFailure::Cancelled)
} else {
Err(StepFailure::TimedOut)
}
}
Err(StepFailure::Io(message)) => Err(StepFailure::Io(message)),
}
}
async fn run_streaming(
mut command: Command,
stdin: Option<Vec<u8>>,
timeout_secs: &u64,
hooks: StepHooks,
) -> anyhow::Result<()> {
let _ = stdin;
let mut child = command.spawn()?;
let stdout = child.stdout.take();
let stderr = child.stderr.take();
let stdout_hook = Arc::clone(&hooks.on_log);
let stdout_task = stdout.map(|stdout| {
tokio::spawn(async move {
let mut lines = BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
stdout_hook(format!("{line}\n"));
}
})
});
let stderr_hook = Arc::clone(&hooks.on_log);
let stderr_task = stderr.map(|stderr| {
tokio::spawn(async move {
let mut lines = BufReader::new(stderr).lines();
while let Ok(Some(line)) = lines.next_line().await {
stderr_hook(format!("{line}\n"));
}
})
});
let status = tokio::time::timeout(
std::time::Duration::from_secs((*timeout_secs).max(1)),
child.wait(),
)
.await;
if let Some(task) = stdout_task {
let _ = task.await;
}
if let Some(task) = stderr_task {
let _ = task.await;
}
match status {
Ok(status) => {
let status = status?;
if status.success() {
Ok(())
} else {
anyhow::bail!("command exited with {:?}", status.code())
}
}
Err(_) => {
let _ = child.kill().await;
anyhow::bail!("command timed out")
}
}
}
async fn collect_artifacts(
_cfg: &RunnerConfig,
payload: &JobPayload,
repo_dir: &Path,
on_artifact: std::sync::Arc<dyn Fn(String, String, i64, Vec<u8>) + Send + Sync>,
) {
for spec in &payload.artifacts {
let base = repo_dir.join(&spec.path);
let mut files: Vec<(String, Vec<u8>)> = Vec::new();
if base.is_file() {
if let Ok(bytes) = std::fs::read(&base) {
files.push((spec.path.clone(), bytes));
}
} else if base.is_dir() {
collect_dir(&base, &spec.path, &mut files);
} else if let Ok(set) = globset::Glob::new(&spec.path) {
let matcher = set.compile_matcher();
walk(repo_dir, repo_dir, &mut |rel, path| {
if matcher.is_match(rel)
&& let Ok(bytes) = std::fs::read(path)
{
files.push((rel.to_string(), bytes));
}
});
}
if files.is_empty() {
continue;
}
// A spec maps to a single artifact; directory/glob payloads are
// tar-less concatenations preserved per file under the artifact name.
if files.len() == 1 {
let (_, bytes) = files.remove(0);
let size = bytes.len() as i64;
on_artifact(spec.name.clone(), spec.path.clone(), size, bytes);
} else {
let mut bytes = Vec::new();
for (rel, content) in files {
bytes.extend_from_slice(format!("--- {rel} ---\n").as_bytes());
bytes.extend_from_slice(&content);
bytes.push(b'\n');
}
let size = bytes.len() as i64;
on_artifact(spec.name.clone(), spec.path.clone(), size, bytes);
}
}
}
fn collect_dir(dir: &Path, prefix: &str, out: &mut Vec<(String, Vec<u8>)>) {
walk(dir, dir, &mut |rel, path| {
if path.is_file()
&& let Ok(bytes) = std::fs::read(path)
{
out.push((format!("{prefix}/{rel}"), bytes));
}
});
}
fn walk(root: &Path, dir: &Path, visit: &mut dyn FnMut(&str, &Path)) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
walk(root, &path, visit);
} else if let Ok(rel) = path.strip_prefix(root) {
visit(&rel.to_string_lossy(), &path);
}
}
}
fn cleanup(cfg: &RunnerConfig, workspace: &Path) {
if cfg.keep_workspace {
return;
}
let _ = std::fs::remove_dir_all(workspace);
}
pub async fn upload_file(url: &str, bytes: Vec<u8>) -> anyhow::Result<()> {
let span = export::telemetry::http_client_span("PUT", url);
let response =
export::telemetry::send(span, reqwest::Client::new().put(url).body(bytes)).await?;
if !response.status().is_success() {
anyhow::bail!("upload failed with {}", response.status());
}
Ok(())
}
#[must_use]
pub fn sha256_hex(bytes: &[u8]) -> String {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(bytes);
let mut out = String::with_capacity(digest.len() * 2);
for byte in digest {
out.push_str(&format!("{byte:02x}"));
}
out
}