Menu
AkurAI-Build
publicLatest change 0f97f2565890fc5015cf9cc035532809caec56cb - feat(pr): close, reopen, edit and comment on pull requests; delete branches by Ólafur Búi Ólafsson
use std::{
collections::{BTreeMap, BTreeSet},
env,
fs::{self, File, OpenOptions},
io::{Read, Write},
path::{Path, PathBuf},
process::{Command, ExitStatus, Stdio},
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
},
thread,
time::{Duration, Instant},
};
use anyhow::{Context, Result, bail, ensure};
use serde::Serialize;
use sha2::{Digest, Sha256};
use crate::{
config::{self, JobSpec},
db::{Artifact, Database, ManagedDeployment, RunDetail},
};
const MAX_LOG_BYTES: usize = 2 * 1024 * 1024;
const MAX_ARTIFACT_FILES: usize = 10_000;
const MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024;
const CANCEL_GRACE: Duration = Duration::from_secs(10);
#[derive(Clone)]
struct RunCancellation {
requested: Arc<AtomicBool>,
process: Arc<Mutex<Option<ProcessControl>>>,
stop_sent: Arc<AtomicBool>,
force_sent: Arc<AtomicBool>,
}
impl RunCancellation {
fn new() -> Self {
Self {
requested: Arc::new(AtomicBool::new(false)),
process: Arc::new(Mutex::new(None)),
stop_sent: Arc::new(AtomicBool::new(false)),
force_sent: Arc::new(AtomicBool::new(false)),
}
}
fn is_requested(&self) -> bool {
self.requested.load(Ordering::Acquire)
}
fn attach(&self, process: ProcessControl) {
if let Ok(mut current) = self.process.lock() {
*current = Some(process);
}
self.stop_sent.store(false, Ordering::Release);
self.stop_active();
}
fn detach(&self) {
if let Ok(mut current) = self.process.lock() {
*current = None;
}
}
fn request(&self) {
self.requested.store(true, Ordering::Release);
self.stop_active();
}
fn stop_active(&self) {
if !self.is_requested() || self.stop_sent.swap(true, Ordering::AcqRel) {
return;
}
let process = self.process.lock().ok().and_then(|current| current.clone());
if let Some(process) = process {
let _ = process.stop();
}
}
fn force_active(&self) -> Result<()> {
if !self.is_requested() || self.force_sent.swap(true, Ordering::AcqRel) {
return Ok(());
}
let process = self.process.lock().ok().and_then(|current| current.clone());
if let Some(process) = process {
process.force()?;
}
Ok(())
}
}
#[derive(Clone)]
enum ProcessKind {
Native,
Docker(String),
}
#[derive(Clone)]
struct ProcessControl {
pid: u32,
kind: ProcessKind,
}
impl ProcessControl {
fn stop(&self) -> Result<()> {
match &self.kind {
ProcessKind::Native => signal_process_group(self.pid, "TERM"),
ProcessKind::Docker(name) => {
let _ = signal_process_group(self.pid, "TERM");
stop_docker_container(name)
}
}
}
fn force(&self) -> Result<()> {
match &self.kind {
ProcessKind::Native => signal_process_group(self.pid, "KILL"),
ProcessKind::Docker(name) => {
let output = Command::new("docker")
.args(["kill", name])
.output()
.context("force-stop canceled Docker container")?;
ensure!(
output.status.success()
|| String::from_utf8_lossy(&output.stderr).contains("No such container"),
"docker kill failed: {}",
bounded(&String::from_utf8_lossy(&output.stderr), 16 * 1024)
);
Ok(())
}
}
}
}
fn stop_docker_container(name: &str) -> Result<()> {
let output = Command::new("docker")
.args(["stop", "--time", &CANCEL_GRACE.as_secs().to_string(), name])
.output()
.context("stop canceled Docker container")?;
ensure!(
output.status.success()
|| String::from_utf8_lossy(&output.stderr).contains("No such container"),
"docker stop failed: {}",
bounded(&String::from_utf8_lossy(&output.stderr), 16 * 1024)
);
Ok(())
}
#[cfg(unix)]
fn signal_process_group(pid: u32, signal: &str) -> Result<()> {
ensure!(pid > 1, "refusing to signal invalid process group");
let stat = fs::read_to_string(format!("/proc/{pid}/stat"))
.with_context(|| format!("read process identity for {pid}"))?;
let fields = stat
.rsplit_once(')')
.context("invalid process identity")?
.1
.split_whitespace()
.collect::<Vec<_>>();
let process_group = fields.get(2).and_then(|value| value.parse::<u64>().ok());
let session = fields.get(3).and_then(|value| value.parse::<u64>().ok());
ensure!(
process_group == Some(u64::from(pid)) && session == Some(u64::from(pid)),
"refusing to signal unverified process group {pid}"
);
let target = format!("-{pid}");
let status = Command::new("kill")
.args([format!("-{signal}"), "--".to_owned(), target])
.status()
.context("signal process group")?;
ensure!(status.success(), "process-group signal failed for {pid}");
Ok(())
}
#[cfg(not(unix))]
fn signal_process_group(_pid: u32, _signal: &str) -> Result<()> {
bail!("process-tree cancellation is unsupported on this platform")
}
#[derive(Clone)]
pub struct Runner {
database: Database,
root: PathBuf,
allow_native: bool,
cancellations: Arc<Mutex<BTreeMap<i64, Arc<RunCancellation>>>>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct QueuedRun {
pub id: i64,
pub deduplicated: bool,
}
#[derive(Serialize)]
struct DeploymentArtifact<'a> {
id: i64,
name: &'a str,
sha256: &'a str,
bytes: i64,
}
fn sha256_hex(value: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(value);
format!("{:x}", hasher.finalize())
}
struct ProcessResult {
status: ExitStatus,
logs: String,
timed_out: bool,
canceled: bool,
}
impl Runner {
pub fn new(database: Database, root: PathBuf, allow_native: bool) -> Result<Self> {
for path in [
root.join("workspaces"),
root.join("artifacts"),
root.join("cache"),
] {
fs::create_dir_all(&path)?;
private_directory(&path)?;
}
Ok(Self {
database,
root,
allow_native,
cancellations: Arc::new(Mutex::new(BTreeMap::new())),
})
}
/// Request cancellation through the worker that owns a running run.
/// Queued and waiting runs retain the database-only cancellation path.
pub fn cancel_run(&self, id: i64) -> Result<RunDetail> {
let request_deadline = Instant::now() + Duration::from_secs(5);
let cancellation = loop {
let run = self.database.run(id)?;
match run.status.as_str() {
"queued" | "waiting" => {
self.database.cancel_run(id)?;
return self.database.detail(id);
}
"running" => {
let cancellation = self
.cancellations
.lock()
.ok()
.and_then(|active| active.get(&id).cloned());
if let Some(cancellation) = cancellation {
break cancellation;
}
ensure!(
Instant::now() < request_deadline,
"running run {id} has no active worker cancellation control"
);
}
"succeeded" | "failed" | "canceled" | "interrupted" => {
bail!(
"run {id} is {}; only queued, waiting, or running runs can be canceled",
run.status
)
}
status => bail!("run {id} has invalid status {status}"),
}
thread::sleep(Duration::from_millis(50));
};
cancellation.request();
let finish_deadline = Instant::now() + CANCEL_GRACE + Duration::from_secs(5);
loop {
let detail = self.database.detail(id)?;
if matches!(
detail.run.status.as_str(),
"succeeded" | "failed" | "canceled" | "interrupted"
) {
return Ok(detail);
}
ensure!(
Instant::now() < finish_deadline,
"run {id} cancellation did not settle before the worker deadline"
);
thread::sleep(Duration::from_millis(100));
}
}
pub fn queue(
&self,
repository: &str,
git_ref: Option<&str>,
commit: Option<&str>,
trigger: &str,
) -> Result<QueuedRun> {
let repository = self.database.repository(repository)?;
let git_ref = git_ref.unwrap_or(&repository.default_branch);
config::validate_ref(git_ref)?;
let queued = match commit {
Some(commit) => {
config::validate_commit(commit)?;
let (id, deduplicated) = self.database.create_or_get_active_run(
repository.id,
git_ref,
commit,
trigger,
)?;
QueuedRun { id, deduplicated }
}
None => {
let (id, deduplicated) =
self.database
.create_or_get_pending_run(repository.id, git_ref, trigger)?;
QueuedRun { id, deduplicated }
}
};
if !queued.deduplicated {
self.database
.supersede_waiting_runs(repository.id, git_ref, queued.id)?;
}
Ok(queued)
}
/// Delete workspaces of terminal runs older than the retention window.
/// Successful runs already clean up after themselves; this reaps failed,
/// canceled, and interrupted debris plus workspaces of deleted runs.
/// Non-numeric entries (shared repo checkouts) are never touched.
pub fn prune_stale_workspaces(&self) -> Result<usize> {
const RETENTION_SECS: i64 = 7 * 24 * 60 * 60;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0);
let mut removed = 0;
for entry in fs::read_dir(self.root.join("workspaces"))? {
let entry = entry?;
let Some(run_id) = entry
.file_name()
.to_str()
.and_then(|name| name.parse::<i64>().ok())
else {
continue;
};
let stale = match self.database.run(run_id) {
Ok(run) => {
matches!(
run.status.as_str(),
"succeeded" | "failed" | "canceled" | "interrupted"
) && run.finished_at.is_some_and(|at| now - at > RETENTION_SECS)
}
// Run row gone (pruned from the DB): workspace is orphaned.
Err(_) => true,
};
if stale && fs::remove_dir_all(entry.path()).is_ok() {
removed += 1;
}
}
Ok(removed)
}
pub fn process(&self, run_id: i64) -> Result<RunDetail> {
if !self.database.claim_run(run_id)? {
return self.database.detail(run_id);
}
self.process_claimed_run(run_id)
}
pub fn process_claimed_run(&self, run_id: i64) -> Result<RunDetail> {
ensure!(
self.database.run(run_id)?.status == "running",
"run {run_id} was not claimed"
);
let cancellation = Arc::new(RunCancellation::new());
if let Ok(mut active) = self.cancellations.lock() {
active.insert(run_id, cancellation.clone());
}
let result = self.execute_claimed(run_id, &cancellation);
if let Err(error) = result {
if cancellation.is_requested() {
self.database.cancel_running_run(run_id)?;
} else {
let message = bounded(&format!("{error:#}"), 16 * 1024);
self.database.finish_run(run_id, "failed", Some(&message))?;
}
}
if let Ok(mut active) = self.cancellations.lock() {
active.remove(&run_id);
}
self.database.detail(run_id)
}
fn execute_claimed(&self, run_id: i64, cancellation: &RunCancellation) -> Result<()> {
let run = self.database.run(run_id)?;
let repository = self.database.repository(&run.repository)?;
let workspace = self.root.join("workspaces").join(run_id.to_string());
if self.database.jobs(run_id)?.is_empty() {
if workspace.exists() {
fs::remove_dir_all(&workspace)?;
}
fs::create_dir_all(&workspace)?;
let commit = checkout(
&repository.url,
&run.git_ref,
run.commit_sha.as_deref(),
&workspace,
)?;
self.database.set_run_commit(run_id, &commit)?;
let pipeline_path = [workspace.join(".akurai.yml"), workspace.join("akurai.yml")]
.into_iter()
.find(|path| path.is_file())
.context("repository has no .akurai.yml")?;
ensure!(
pipeline_path.metadata()?.len() <= 256 * 1024,
"pipeline is too large"
);
let source = fs::read_to_string(pipeline_path)?;
let jobs = config::parse(&source)?;
self.database.insert_jobs(run_id, &jobs)?;
}
loop {
if cancellation.is_requested() {
self.database.cancel_running_run(run_id)?;
return Ok(());
}
let jobs = self.database.jobs(run_id)?;
if jobs.iter().all(|job| {
matches!(
job.status.as_str(),
"succeeded" | "failed" | "skipped" | "canceled" | "interrupted"
)
}) {
if cancellation.is_requested() {
self.database.cancel_running_run(run_id)?;
return Ok(());
}
let canceled = jobs.iter().any(|job| job.status == "canceled");
let failed = jobs
.iter()
.any(|job| matches!(job.status.as_str(), "failed" | "interrupted"));
self.database.finish_run(
run_id,
if canceled {
"canceled"
} else if failed {
"failed"
} else {
"succeeded"
},
None,
)?;
if !canceled && !failed {
fs::remove_dir_all(&workspace).with_context(|| {
format!("remove completed run workspace {}", workspace.display())
})?;
}
return Ok(());
}
let mut progressed = false;
for job in &jobs {
if cancellation.is_requested() {
self.database.cancel_running_run(run_id)?;
return Ok(());
}
if job.status != "queued" {
continue;
}
let dependencies = jobs
.iter()
.filter(|candidate| job.needs.contains(&candidate.base_name))
.collect::<Vec<_>>();
if dependencies.iter().any(|dependency| {
matches!(
dependency.status.as_str(),
"failed" | "skipped" | "canceled" | "interrupted"
)
}) {
self.database.set_job_status(
job.id,
"skipped",
Some("dependency did not succeed"),
None,
)?;
progressed = true;
continue;
}
if !dependencies
.iter()
.all(|dependency| dependency.status == "succeeded")
{
continue;
}
if job.approval_required && job.approved_at.is_none() {
self.database
.set_job_status(job.id, "waiting", None, None)?;
progressed = true;
continue;
}
self.execute_job(run_id, job.id, &workspace, cancellation)?;
progressed = true;
break;
}
if !progressed {
if cancellation.is_requested() {
self.database.cancel_running_run(run_id)?;
return Ok(());
}
let waiting = self
.database
.jobs(run_id)?
.iter()
.any(|job| job.status == "waiting");
if waiting {
self.database.finish_run(run_id, "waiting", None)?;
return Ok(());
}
bail!("pipeline made no progress");
}
}
}
fn execute_job(
&self,
run_id: i64,
job_id: i64,
workspace: &Path,
cancellation: &RunCancellation,
) -> Result<()> {
let run = self.database.run(run_id)?;
let jobs = self.database.jobs(run_id)?;
let job = jobs
.iter()
.find(|job| job.id == job_id)
.context("job disappeared")?;
let spec: JobSpec = serde_json::from_str(&job.spec_json)?;
let branch = run
.git_ref
.strip_prefix("refs/heads/")
.unwrap_or(&run.git_ref);
if !spec.branches.is_empty() && !spec.branches.iter().any(|allowed| allowed == branch) {
self.database.set_job_status(
job_id,
"skipped",
Some("Git ref is not allowed for this job"),
None,
)?;
return Ok(());
}
let repository_trusted = self.database.repository_trusted(run.repository_id)?;
let repository_default_branch = self.database.repository(&run.repository)?.default_branch;
// A trusted *repository* only earns unrestricted execution on its
// default branch. PR/feature branches on trusted repos still run
// through the untrusted-docker pinned-image path so untrusted branch
// code never gets native/trusted-docker deploy-staging access.
// A merge-queue staging ref is the exception: it is an independently
// approved merge commit that the queue is about to land on the
// default branch, and CI on it is what gates that landing.
let queue_targets_default = run
.git_ref
.strip_prefix("refs/merge-queue/")
.and_then(|id| id.parse::<i64>().ok())
.and_then(|id| self.database.merge_queue_entry(id).ok())
.filter(|entry| entry.repository_id == run.repository_id)
.and_then(|entry| {
self.database
.pull_request(&entry.repository, entry.pull_request_number)
.ok()
})
.is_some_and(|detail| detail.pull_request.target_ref == repository_default_branch);
let ref_trusted =
repository_trusted && (branch == repository_default_branch || queue_targets_default);
if spec.executor == "native" && (!self.allow_native || !ref_trusted) {
self.database.set_job_status(
job_id,
"failed",
Some(
"native execution requires a trusted repository, its default branch (or an approved refs/merge-queue/N ref), and controller native opt-in. For CI on a feature branch, give the job `executor: docker` with a digest-pinned image (name@sha256:...): untrusted branch code runs there without native/trusted-docker host access.",
),
None,
)?;
return Ok(());
}
if spec.executor == "docker"
&& !ref_trusted
&& !spec
.image
.as_deref()
.is_some_and(config::is_immutable_image)
{
self.database.set_job_status(
job_id,
"failed",
Some("untrusted repositories require an immutable sha256 Docker image digest"),
None,
)?;
return Ok(());
}
if !ref_trusted && (!spec.cache.is_empty() || spec.environment.is_some()) {
self.database.set_job_status(
job_id,
"failed",
Some("untrusted refs cannot use shared caches or deployment environments"),
None,
)?;
return Ok(());
}
let artifacts = self.database.artifacts(run_id)?;
let dependency_ids = jobs
.iter()
.filter(|candidate| spec.needs.contains(&candidate.base_name))
.map(|candidate| candidate.id)
.collect::<BTreeSet<_>>();
let promoted = artifacts
.iter()
.filter(|artifact| dependency_ids.contains(&artifact.job_id))
.collect::<Vec<_>>();
let (deployment, rollback_of, recovery_of) = if let Some(environment) = &spec.environment {
let manifest = promoted
.iter()
.map(|artifact| DeploymentArtifact {
id: artifact.id,
name: &artifact.name,
sha256: &artifact.sha256,
bytes: artifact.bytes,
})
.collect::<Vec<_>>();
let artifact_manifest = serde_json::to_string(&manifest)?;
let artifact_digest = sha256_hex(artifact_manifest.as_bytes());
let managed = match (&spec.installation_id, spec.installation_revision) {
(Some(installation_id), Some(installation_revision)) => {
Some((installation_id.as_str(), installation_revision))
}
(None, None) => None,
_ => bail!("managed deployment requires installation_id and installation_revision"),
};
let deployment = if let Some((installation_id, installation_revision)) = managed {
self.database.begin_managed_deployment(ManagedDeployment {
run_id,
job_id,
environment,
artifacts_json: &artifact_manifest,
installation_id,
installation_revision,
commit_sha: run.commit_sha.as_deref(),
artifact_digest: Some(&artifact_digest),
result: None,
rollback_of_deployment_id: spec.rollback_of_deployment_id,
recovery_of_deployment_id: spec.recovery_of_deployment_id,
})?
} else {
self.database
.begin_deployment(run_id, job_id, environment, &artifact_manifest)?
};
(
Some(deployment),
spec.rollback_of_deployment_id,
spec.recovery_of_deployment_id,
)
} else {
(None, None, None)
};
let deployment_result = |success: bool| {
if !success {
"failure"
} else if recovery_of.is_some() {
"recovery"
} else if rollback_of.is_some() {
"rollback"
} else {
"success"
}
};
self.database
.set_job_status(job_id, "running", None, None)?;
let execution = self.execute(&spec, workspace, run_id, job_id, cancellation);
match execution {
Ok(result) if cancellation.is_requested() || result.canceled => {
let logs = bounded(
&format!("{}\njob canceled by operator", result.logs),
MAX_LOG_BYTES,
);
self.database.set_job_status(
job_id,
"canceled",
Some(&logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment_result(
id,
"failed",
Some(deployment_result(false)),
)?;
}
}
Ok(result) if result.status.success() && !result.timed_out => {
if let Err(error) =
self.collect_artifacts(run_id, job_id, workspace, &spec.artifacts)
{
let logs = format!("{}\nartifact error: {error:#}", result.logs);
self.database.set_job_status(
job_id,
"failed",
Some(&logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment_result(
id,
"failed",
Some(deployment_result(false)),
)?;
}
} else {
self.database.set_job_status(
job_id,
"succeeded",
Some(&result.logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment_result(
id,
"succeeded",
Some(deployment_result(true)),
)?;
}
}
}
Ok(result) => {
let logs = if result.timed_out {
format!("{}\njob timed out", result.logs)
} else {
result.logs
};
self.database.set_job_status(
job_id,
"failed",
Some(&logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment_result(
id,
"failed",
Some(deployment_result(false)),
)?;
}
}
Err(error) if cancellation.is_requested() => {
let logs = bounded(
&format!("job canceled by operator: {error:#}"),
MAX_LOG_BYTES,
);
self.database
.set_job_status(job_id, "canceled", Some(&logs), None)?;
if let Some(id) = deployment {
self.database.finish_deployment_result(
id,
"failed",
Some(deployment_result(false)),
)?;
}
}
Err(error) => {
self.database.set_job_status(
job_id,
"failed",
Some(&bounded(&format!("{error:#}"), MAX_LOG_BYTES)),
None,
)?;
if let Some(id) = deployment {
self.database.finish_deployment_result(
id,
"failed",
Some(deployment_result(false)),
)?;
}
}
}
Ok(())
}
/// Build-context variables injected into every job, native or docker, so
/// pipelines never hardcode a worker username or data root. `workspace` is
/// the path the job actually sees, which differs between executors.
fn build_context(&self, workspace: &str) -> BTreeMap<String, String> {
let mut context = BTreeMap::new();
context.insert(
"AKURAI_BUILD_ROOT".to_owned(),
self.root.display().to_string(),
);
context.insert(
"AKURAI_BUILD_HOSTED".to_owned(),
self.root.join("hosted").display().to_string(),
);
context.insert("AKURAI_BUILD_WORKSPACE".to_owned(), workspace.to_owned());
// The invoking user's home, captured before HOME is repointed at the
// per-job sandbox, so pipelines can reach shared toolchain caches.
if let Some(home) = env::var_os("HOME").and_then(|home| home.into_string().ok()) {
context.insert("AKURAI_BUILD_HOME".to_owned(), home);
}
context
}
fn execute(
&self,
spec: &JobSpec,
workspace: &Path,
run_id: i64,
job_id: i64,
cancellation: &RunCancellation,
) -> Result<ProcessResult> {
let secrets = declared_secrets(&spec.secrets)?;
let matrix = matrix_environment(&spec.matrix);
if spec.executor == "native" {
let context = self.build_context(&workspace.display().to_string());
let repository_id = if spec.cache.is_empty() {
None
} else {
Some(self.database.run(run_id)?.repository_id)
};
if let Some(repository_id) = repository_id {
self.prepare_native_cache(repository_id, workspace, &spec.cache)?;
}
let mut command = native_shell(spec.shell.as_deref(), &spec.command);
command.current_dir(workspace).env_clear();
copy_host_environment(&mut command);
command.env("HOME", workspace.join(".home"));
for (name, value) in context.iter().chain(&matrix).chain(&secrets) {
command.env(name, value);
}
let result = run_process(
command,
spec.timeout,
&secrets,
ProcessKind::Native,
cancellation,
);
if let Some(repository_id) = repository_id {
self.persist_native_cache(repository_id, workspace, &spec.cache)?;
}
result
} else {
self.execute_docker(spec, workspace, run_id, job_id, &secrets, cancellation)
}
}
fn execute_docker(
&self,
spec: &JobSpec,
workspace: &Path,
run_id: i64,
job_id: i64,
secrets: &BTreeMap<String, String>,
cancellation: &RunCancellation,
) -> Result<ProcessResult> {
let matrix = matrix_environment(&spec.matrix);
let name = format!("akurai-{run_id}-{job_id}");
cleanup_orphan_container(&name)?;
let mut command = grouped_command("docker");
command.args([
"run",
"--rm",
"--name",
&name,
"--pull=missing",
"--cap-drop=ALL",
"--security-opt=no-new-privileges",
"--pids-limit=512",
"--memory=4g",
"--cpus=2",
"--read-only",
"--tmpfs=/tmp:rw,nosuid,nodev,size=512m",
"--workdir=/workspace",
]);
if !spec.network {
command.arg("--network=none");
}
if let Some(platform) = spec.matrix.get("platform") {
command.args(["--platform", platform]);
}
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
let metadata = workspace.metadata()?;
command.args(["--user", &format!("{}:{}", metadata.uid(), metadata.gid())]);
}
command.args(["--volume", &format!("{}:/workspace", workspace.display())]);
let repository_id = self.database.run(run_id)?.repository_id;
for path in &spec.cache {
let (host, target) = self.cache_paths(repository_id, workspace, path)?;
command.args([
"--volume",
&format!("{}:/workspace/{}", host.display(), path),
]);
ensure!(target.is_dir(), "cache target is not a directory");
}
command.args(["--env", "HOME=/workspace/.home"]);
// Inside the container the job's own tree is always /workspace; the
// root/hosted/home values remain the host-side truth.
for (name, value) in self.build_context("/workspace") {
command.args(["--env", &format!("{name}={value}")]);
}
for name in matrix.keys().chain(secrets.keys()) {
command.args(["--env", name]);
}
let image = spec.image.as_deref().context("docker job has no image")?;
command.arg(image);
match spec.shell.as_deref().unwrap_or("sh") {
"cmd" => command.args(["cmd.exe", "/d", "/s", "/c", &spec.command]),
"powershell" => command.args([
"powershell",
"-NoProfile",
"-NonInteractive",
"-Command",
&spec.command,
]),
"sh" => command.args(["sh", "-eu", "-c", &spec.command]),
other => bail!("unsupported shell {other}"),
};
command.env_clear();
copy_host_environment(&mut command);
for (key, value) in matrix.iter().chain(secrets) {
command.env(key, value);
}
let mut result = run_process(
command,
spec.timeout,
secrets,
ProcessKind::Docker(name.clone()),
cancellation,
)?;
if result.timed_out || result.canceled {
match Command::new("docker")
.args(["rm", "--force", &name])
.output()
{
Ok(output) if !output.status.success() => result.logs.push_str(&format!(
"\nfailed to remove stopped container {name}: {}",
String::from_utf8_lossy(&output.stderr)
)),
Err(error) => result.logs.push_str(&format!(
"\nfailed to remove stopped container {name}: {error}"
)),
Ok(_) => {}
}
}
Ok(result)
}
fn cache_paths(
&self,
repository_id: i64,
workspace: &Path,
path: &str,
) -> Result<(PathBuf, PathBuf)> {
ensure!(repository_id > 0, "invalid repository id for cache");
validate_cache_path(path)?;
let cache_root = self.root.join("cache").join(repository_id.to_string());
fs::create_dir_all(&cache_root)?;
reject_symlink(&cache_root)?;
let host = cache_root.join(cache_component(path));
let target = workspace.join(path);
create_secure_directory(&host)?;
create_secure_directory(&target)?;
let canonical_root = cache_root.canonicalize()?;
ensure!(
host.canonicalize()?.starts_with(&canonical_root),
"cache path escapes repository cache root"
);
let canonical_workspace = workspace.canonicalize()?;
ensure!(
target.canonicalize()?.starts_with(&canonical_workspace),
"cache target escapes workspace"
);
Ok((host, target))
}
fn prepare_native_cache(
&self,
repository_id: i64,
workspace: &Path,
paths: &[String],
) -> Result<()> {
for path in paths {
let (host, target) = self.cache_paths(repository_id, workspace, path)?;
clear_directory(&target)?;
copy_tree(&host, &target)?;
}
Ok(())
}
fn persist_native_cache(
&self,
repository_id: i64,
workspace: &Path,
paths: &[String],
) -> Result<()> {
for path in paths {
let (host, target) = self.cache_paths(repository_id, workspace, path)?;
clear_directory(&host)?;
copy_tree(&target, &host)?;
}
Ok(())
}
fn collect_artifacts(
&self,
run_id: i64,
job_id: i64,
workspace: &Path,
patterns: &[String],
) -> Result<()> {
if patterns.is_empty() {
return Ok(());
}
let workspace = workspace.canonicalize()?;
let artifact_root = self.root.join("artifacts");
fs::create_dir_all(&artifact_root)?;
let existing = self.database.artifacts(run_id)?;
let mut bytes = existing.iter().try_fold(0_u64, |total, artifact| {
Ok::<_, anyhow::Error>(total + u64::try_from(artifact.bytes)?)
})?;
let mut count = existing.len();
let mut files = BTreeSet::new();
for pattern in patterns {
let full = workspace.join(pattern).to_string_lossy().into_owned();
for matched in
glob::glob(&full).with_context(|| format!("invalid artifact glob {pattern}"))?
{
collect_files(&matched?, &mut files)?;
}
}
for source in files {
let canonical = source.canonicalize()?;
ensure!(
canonical.starts_with(&workspace),
"artifact escapes workspace"
);
ensure!(
!source.symlink_metadata()?.file_type().is_symlink(),
"artifact symlinks are not allowed"
);
let relative = canonical.strip_prefix(&workspace)?;
let size = canonical.metadata()?.len();
bytes = bytes.checked_add(size).context("artifact size overflow")?;
count += 1;
ensure!(
bytes <= MAX_ARTIFACT_BYTES && count <= MAX_ARTIFACT_FILES,
"artifact limits exceeded"
);
let relative_store = PathBuf::from("artifacts")
.join(run_id.to_string())
.join(job_id.to_string())
.join(relative);
let destination = self.root.join(&relative_store);
let parent = destination.parent().context("artifact has no parent")?;
fs::create_dir_all(parent)?;
let temporary =
destination.with_extension(format!("akurai-tmp-{}", std::process::id()));
let (digest, copied) = copy_hashed(&canonical, &temporary)?;
fs::rename(&temporary, &destination)?;
self.database.add_artifact(
run_id,
job_id,
&relative.to_string_lossy(),
&relative_store.to_string_lossy(),
&digest,
copied,
)?;
}
Ok(())
}
pub fn artifact_path(&self, artifact: &Artifact) -> Result<PathBuf> {
let root = self.root.canonicalize()?;
let path = self.root.join(&artifact.path).canonicalize()?;
ensure!(
path.starts_with(&root) && path.is_file(),
"artifact path is invalid"
);
let (digest, bytes) = hash_file(&path)?;
ensure!(
digest == artifact.sha256 && i64::try_from(bytes)? == artifact.bytes,
"artifact integrity check failed"
);
Ok(path)
}
}
fn cleanup_orphan_container(name: &str) -> Result<()> {
let output = Command::new("docker")
.args(["rm", "--force", name])
.output()
.context("clean up Docker container")?;
if output.status.success() {
return Ok(());
}
let error = String::from_utf8_lossy(&output.stderr);
ensure!(
error.to_ascii_lowercase().contains("no such container"),
"failed to clean up Docker container {name}: {}",
bounded(&error, 4096)
);
Ok(())
}
fn validate_cache_path(path: &str) -> Result<()> {
ensure!(
!path.is_empty() && !path.contains(['\\', ':', '\0', '*', '?', '[', ']']),
"invalid cache path"
);
ensure!(
Path::new(path)
.components()
.all(|component| matches!(component, std::path::Component::Normal(_))),
"cache path must stay inside the workspace"
);
Ok(())
}
fn reject_symlink(path: &Path) -> Result<()> {
if let Ok(metadata) = path.symlink_metadata() {
ensure!(
!metadata.file_type().is_symlink(),
"cache symlinks are not allowed"
);
}
Ok(())
}
fn create_secure_directory(path: &Path) -> Result<()> {
let mut current = PathBuf::new();
for component in path.components() {
current.push(component);
if current.exists() {
reject_symlink(¤t)?;
ensure!(current.is_dir(), "{} is not a directory", current.display());
} else {
fs::create_dir(¤t)?;
}
}
Ok(())
}
fn clear_directory(path: &Path) -> Result<()> {
reject_symlink(path)?;
ensure!(path.is_dir(), "{} is not a directory", path.display());
for entry in fs::read_dir(path)? {
let entry = entry?;
let child = entry.path();
reject_symlink(&child)?;
if child.is_dir() {
fs::remove_dir_all(child)?;
} else {
fs::remove_file(child)?;
}
}
Ok(())
}
fn copy_tree(source: &Path, destination: &Path) -> Result<()> {
if !source.exists() {
return Ok(());
}
reject_symlink(source)?;
let metadata = source.metadata()?;
if metadata.is_dir() {
create_secure_directory(destination)?;
for entry in fs::read_dir(source)? {
let entry = entry?;
copy_tree(&entry.path(), &destination.join(entry.file_name()))?;
}
} else if metadata.is_file() {
if let Some(parent) = destination.parent() {
create_secure_directory(parent)?;
}
reject_symlink(destination)?;
fs::copy(source, destination)?;
} else {
bail!("cache entry is not a regular file or directory");
}
Ok(())
}
fn checkout(url: &str, git_ref: &str, requested: Option<&str>, workspace: &Path) -> Result<String> {
run_git(workspace, ["init"])?;
run_git(workspace, ["remote", "add", "origin", url])?;
run_git(workspace, ["fetch", "--depth=64", "origin", git_ref])?;
let branch_tip = git_output(workspace, ["rev-parse", "FETCH_HEAD"])?;
let commit = if let Some(commit) = requested {
run_git(workspace, ["fetch", "--depth=64", "origin", commit])?;
let status = git_status(
workspace,
["merge-base", "--is-ancestor", commit, &branch_tip],
)?;
ensure!(
status.success(),
"requested commit is not reachable from fetched ref"
);
commit.to_owned()
} else {
branch_tip
};
run_git(workspace, ["checkout", "--detach", &commit])?;
git_output(workspace, ["rev-parse", "HEAD"])
}
fn run_git<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<()> {
let output = crate::git_process::command(cwd).args(arguments).output()?;
ensure!(
output.status.success(),
"git failed: {}",
bounded(&String::from_utf8_lossy(&output.stderr), 16 * 1024)
);
Ok(())
}
fn git_output<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<String> {
crate::git_process::output(cwd, arguments)
}
fn git_status<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<ExitStatus> {
Ok(crate::git_process::command(cwd).args(arguments).status()?)
}
fn grouped_command(program: &str) -> Command {
#[cfg(unix)]
{
let mut command = Command::new("setsid");
command.args(["--wait", program]);
command
}
#[cfg(not(unix))]
{
Command::new(program)
}
}
fn native_shell(shell: Option<&str>, script: &str) -> Command {
#[cfg(windows)]
{
let mut command = Command::new(match shell.unwrap_or("cmd") {
"powershell" => "powershell",
_ => "cmd.exe",
});
if shell == Some("powershell") {
command.args(["-NoProfile", "-NonInteractive", "-Command", script]);
} else {
command.args(["/d", "/s", "/c", script]);
}
command
}
#[cfg(not(windows))]
{
let _ = shell;
let mut command = grouped_command("/bin/sh");
command.args(["-eu", "-c", script]);
command
}
}
fn run_process(
mut command: Command,
timeout_seconds: u64,
secrets: &BTreeMap<String, String>,
kind: ProcessKind,
cancellation: &RunCancellation,
) -> Result<ProcessResult> {
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = command.spawn().context("start build process")?;
let control = ProcessControl {
pid: child.id(),
kind,
};
cancellation.attach(control.clone());
let stdout = child.stdout.take().context("capture stdout")?;
let stderr = child.stderr.take().context("capture stderr")?;
let out = thread::spawn(move || read_bounded(stdout));
let err = thread::spawn(move || read_bounded(stderr));
let deadline = Instant::now() + Duration::from_secs(timeout_seconds);
let mut cancel_deadline = None;
let (status, timed_out, canceled) = loop {
if let Some(status) = child.try_wait()? {
break (status, false, cancellation.is_requested());
}
if cancellation.is_requested() {
let deadline = cancel_deadline.get_or_insert_with(|| {
cancellation.stop_active();
Instant::now() + CANCEL_GRACE
});
if Instant::now() >= *deadline
&& let Err(error) = cancellation.force_active()
{
child.kill().with_context(|| {
format!("force-stop verified build child after cancellation: {error:#}")
})?;
}
} else if Instant::now() >= deadline {
if let Err(error) = control.force() {
child.kill().with_context(|| {
format!("force-stop verified build child after timeout: {error:#}")
})?;
}
break (child.wait()?, true, false);
}
thread::sleep(Duration::from_millis(100));
};
cancellation.detach();
let stdout = out
.join()
.map_err(|_| anyhow::anyhow!("stdout reader panicked"))??;
let stderr = err
.join()
.map_err(|_| anyhow::anyhow!("stderr reader panicked"))??;
let mut logs = String::from_utf8_lossy(&stdout).into_owned();
if !stderr.is_empty() {
if !logs.is_empty() {
logs.push('\n');
}
logs.push_str(&String::from_utf8_lossy(&stderr));
}
for value in secrets.values().filter(|value| !value.is_empty()) {
logs = logs.replace(value, "[redacted]");
}
Ok(ProcessResult {
status,
logs: bounded(&logs, MAX_LOG_BYTES),
timed_out,
canceled,
})
}
fn read_bounded(mut reader: impl Read) -> Result<Vec<u8>> {
let mut retained = Vec::new();
let mut chunk = [0_u8; 8192];
loop {
let read = reader.read(&mut chunk)?;
if read == 0 {
break;
}
let remaining = MAX_LOG_BYTES.saturating_sub(retained.len());
retained.extend_from_slice(&chunk[..read.min(remaining)]);
}
Ok(retained)
}
fn declared_secrets(names: &[String]) -> Result<BTreeMap<String, String>> {
names
.iter()
.map(|name| {
let value = env::var(format!("AKURAI_SECRET_{name}"))
.with_context(|| format!("missing AKURAI_SECRET_{name}"))?;
Ok((name.clone(), value))
})
.collect()
}
fn matrix_environment(matrix: &BTreeMap<String, String>) -> BTreeMap<String, String> {
matrix
.iter()
.map(|(key, value)| {
let key = key
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() {
character.to_ascii_uppercase()
} else {
'_'
}
})
.collect::<String>();
(format!("AKURAI_MATRIX_{key}"), value.clone())
})
.collect()
}
fn copy_host_environment(command: &mut Command) {
for name in [
"PATH",
"HOME",
"USER",
"LANG",
"LC_ALL",
"DOCKER_CONFIG",
"SSH_AUTH_SOCK",
] {
if let Some(value) = env::var_os(name) {
command.env(name, value);
}
}
}
fn collect_files(path: &Path, files: &mut BTreeSet<PathBuf>) -> Result<()> {
let metadata = path.symlink_metadata()?;
ensure!(
!metadata.file_type().is_symlink(),
"artifact symlinks are not allowed"
);
if metadata.is_file() {
files.insert(path.to_owned());
} else if metadata.is_dir() {
for entry in fs::read_dir(path)? {
collect_files(&entry?.path(), files)?;
}
} else {
bail!("artifact is not a regular file or directory");
}
Ok(())
}
fn copy_hashed(source: &Path, destination: &Path) -> Result<(String, u64)> {
let mut input = File::open(source)?;
let mut output = OpenOptions::new()
.write(true)
.create_new(true)
.open(destination)?;
let mut hasher = Sha256::new();
let mut total = 0_u64;
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = input.read(&mut buffer)?;
if read == 0 {
break;
}
total += u64::try_from(read)?;
ensure!(total <= MAX_ARTIFACT_BYTES, "artifact file is too large");
hasher.update(&buffer[..read]);
output.write_all(&buffer[..read])?;
}
output.sync_all()?;
Ok((format!("{:x}", hasher.finalize()), total))
}
fn hash_file(path: &Path) -> Result<(String, u64)> {
let mut input = File::open(path)?;
let mut hasher = Sha256::new();
let mut total = 0_u64;
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = input.read(&mut buffer)?;
if read == 0 {
break;
}
total += u64::try_from(read)?;
ensure!(total <= MAX_ARTIFACT_BYTES, "artifact file is too large");
hasher.update(&buffer[..read]);
}
Ok((format!("{:x}", hasher.finalize()), total))
}
fn cache_component(path: &str) -> String {
sha256_hex(path.as_bytes())
}
fn private_directory(path: &Path) -> Result<()> {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
}
Ok(())
}
fn bounded(value: &str, max: usize) -> String {
if value.len() <= max {
return value.to_owned();
}
let mut end = max;
while !value.is_char_boundary(end) {
end -= 1;
}
format!("{}\n[truncated]", &value[..end])
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::Database;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
#[test]
fn cache_component_distinguishes_paths_that_sanitize_alike() {
assert_ne!(cache_component("a/b"), cache_component("a_b"));
assert_eq!(cache_component("a/b"), cache_component("a/b"));
}
#[test]
fn cache_paths_reject_workspace_escape_and_globs() {
for path in ["../outside", "/absolute", "a/../../outside", "a/*"] {
assert!(validate_cache_path(path).is_err(), "accepted {path}");
}
assert!(validate_cache_path("target/debug").is_ok());
}
// ── bounded ──────────────────────────────────────────────────────
#[test]
fn bounded_keeps_short_strings() {
assert_eq!(bounded("hello", 100), "hello");
}
#[test]
fn bounded_truncates_long_strings() {
let result = bounded("hello world", 5);
assert!(result.ends_with("[truncated]"));
assert!(result.len() <= "hello[truncated]".len() + 1); // char-boundary lenience
}
#[test]
fn bounded_handles_empty() {
assert_eq!(bounded("", 10), "");
}
#[test]
fn bounded_exact_boundary() {
let input = "12345";
let result = bounded(input, 5);
assert_eq!(result, "12345");
}
// ── matrix_environment ───────────────────────────────────────────
#[test]
fn matrix_environment_converts_keys_to_uppercase() {
let mut matrix = BTreeMap::new();
matrix.insert("platform".to_string(), "linux/amd64".to_string());
let env = matrix_environment(&matrix);
assert_eq!(
env.get("AKURAI_MATRIX_PLATFORM").map(String::as_str),
Some("linux/amd64")
);
}
#[test]
fn matrix_environment_replaces_non_alphanumeric_with_underscore() {
let mut matrix = BTreeMap::new();
matrix.insert("my-key".to_string(), "value".to_string());
let env = matrix_environment(&matrix);
assert_eq!(
env.get("AKURAI_MATRIX_MY_KEY").map(String::as_str),
Some("value")
);
}
#[test]
fn matrix_environment_handles_empty() {
let matrix = BTreeMap::new();
let env = matrix_environment(&matrix);
assert!(env.is_empty());
}
// ── private_directory ────────────────────────────────────────────
#[test]
fn private_directory_creates_and_sets_permissions() -> Result<()> {
let dir = tempfile::tempdir()?;
let sub = dir.path().join("private");
fs::create_dir(&sub)?;
private_directory(&sub)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mode = sub.metadata()?.permissions().mode();
assert_eq!(mode & 0o777, 0o700);
}
Ok(())
}
// ── read_bounded ─────────────────────────────────────────────────
#[test]
fn read_bounded_reads_within_limit() -> Result<()> {
let data = b"hello world";
let result = read_bounded(&data[..])?;
assert_eq!(result, data);
Ok(())
}
#[test]
fn read_bounded_truncates_at_limit() -> Result<()> {
let data = vec![b'x'; MAX_LOG_BYTES + 1024];
let result = read_bounded(&data[..])?;
assert_eq!(result.len(), MAX_LOG_BYTES);
Ok(())
}
#[test]
fn read_bounded_handles_empty() -> Result<()> {
let result = read_bounded(&b""[..])?;
assert!(result.is_empty());
Ok(())
}
// ── collect_files ────────────────────────────────────────────────
#[test]
fn collect_files_gathers_single_file() -> Result<()> {
let dir = tempfile::tempdir()?;
let file = dir.path().join("a.txt");
fs::write(&file, b"content")?;
let mut files = BTreeSet::new();
collect_files(&file, &mut files)?;
assert_eq!(files.len(), 1);
assert!(files.contains(&file));
Ok(())
}
#[test]
fn collect_files_gathers_directory_recursively() -> Result<()> {
let dir = tempfile::tempdir()?;
let sub = dir.path().join("sub");
fs::create_dir(&sub)?;
let a = sub.join("a.txt");
let b = sub.join("b.txt");
fs::write(&a, b"a")?;
fs::write(&b, b"b")?;
let mut files = BTreeSet::new();
collect_files(&sub, &mut files)?;
assert_eq!(files.len(), 2);
assert!(files.contains(&a));
assert!(files.contains(&b));
Ok(())
}
#[test]
fn collect_files_rejects_symlinks() -> Result<()> {
let dir = tempfile::tempdir()?;
let real = dir.path().join("real.txt");
let link = dir.path().join("link.txt");
fs::write(&real, b"content")?;
std::os::unix::fs::symlink(&real, &link)?;
let mut files = BTreeSet::new();
assert!(collect_files(&link, &mut files).is_err());
Ok(())
}
// ── copy_hashed ──────────────────────────────────────────────────
#[test]
fn copy_hashed_copies_and_returns_hash() -> Result<()> {
let dir = tempfile::tempdir()?;
let source = dir.path().join("src.txt");
let dest = dir.path().join("dst.txt");
fs::write(&source, b"hello copy_hashed")?;
let (digest, size) = copy_hashed(&source, &dest)?;
assert_eq!(size, 17);
assert_eq!(digest.len(), 64);
assert!(dest.exists());
assert_eq!(fs::read(&dest)?, b"hello copy_hashed");
Ok(())
}
#[test]
fn copy_hashed_rejects_overwrite() -> Result<()> {
let dir = tempfile::tempdir()?;
let source = dir.path().join("src.txt");
let dest = dir.path().join("dst.txt");
fs::write(&source, b"hello")?;
fs::write(&dest, b"existing")?;
assert!(copy_hashed(&source, &dest).is_err());
Ok(())
}
#[test]
fn copy_hashed_empty_file() -> Result<()> {
let dir = tempfile::tempdir()?;
let source = dir.path().join("src.txt");
let dest = dir.path().join("dst.txt");
fs::write(&source, b"")?;
let (digest, size) = copy_hashed(&source, &dest)?;
assert_eq!(size, 0);
assert_eq!(
digest,
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
);
Ok(())
}
// ── hash_file ────────────────────────────────────────────────────
#[test]
fn hash_file_computes_sha256_and_size() -> Result<()> {
let dir = tempfile::tempdir()?;
let file = dir.path().join("data.txt");
fs::write(&file, b"hash me")?;
let (digest, size) = hash_file(&file)?;
assert_eq!(size, 7);
assert_eq!(digest.len(), 64);
Ok(())
}
#[test]
fn hash_file_deterministic() -> Result<()> {
let dir = tempfile::tempdir()?;
let file = dir.path().join("data.txt");
fs::write(&file, b"same content")?;
let (d1, s1) = hash_file(&file)?;
let (d2, s2) = hash_file(&file)?;
assert_eq!(d1, d2);
assert_eq!(s1, s2);
Ok(())
}
// ── native_shell ─────────────────────────────────────────────────
#[test]
fn native_shell_uses_verified_process_group_on_unix() {
let cmd = native_shell(None, "echo hello");
let program = cmd.get_program().to_string_lossy().into_owned();
#[cfg(not(windows))]
assert_eq!(
program, "setsid",
"expected setsid process wrapper, got {program}"
);
}
#[test]
fn native_shell_passes_script_as_args() {
let cmd = native_shell(None, "echo hello");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert!(
args.contains(&"/bin/sh".to_string()) || args.contains(&"echo hello".to_string()),
"args should contain the shell and script, got {args:?}"
);
}
#[cfg(unix)]
#[test]
fn run_process_cancels_the_verified_process_group() -> Result<()> {
let cancellation = RunCancellation::new();
let worker_cancellation = cancellation.clone();
let handle = thread::spawn(move || {
run_process(
native_shell(None, "sleep 30"),
60,
&BTreeMap::new(),
ProcessKind::Native,
&worker_cancellation,
)
});
thread::sleep(Duration::from_millis(150));
cancellation.request();
let result = handle
.join()
.map_err(|_| anyhow::anyhow!("cancellation worker panicked"))??;
assert!(result.canceled, "process result should record cancellation");
Ok(())
}
// ── Runner construction ───────────────────────────────────────────
#[test]
fn runner_new_creates_required_directories() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let _runner = Runner::new(database, root.path().to_owned(), false)?;
for sub in ["workspaces", "artifacts", "cache"] {
let path = root.path().join(sub);
assert!(path.is_dir(), "{sub} directory should exist");
}
Ok(())
}
#[test]
fn prune_removes_orphans_and_keeps_live_and_shared_workspaces() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let repo = database.add_repository("app", "https://example.com/app.git", "main")?;
// Fresh failed run: terminal but inside the retention window.
let recent = database.create_run(repo.id, "main", None, "manual")?;
assert!(database.claim_run(recent)?);
database.finish_run(recent, "failed", None)?;
let workspaces = root.path().join("workspaces");
fs::create_dir_all(workspaces.join(recent.to_string()))?;
// Orphan: numeric dir with no run row.
fs::create_dir_all(workspaces.join("999999"))?;
// Shared checkout: non-numeric, never touched.
fs::create_dir_all(workspaces.join("SharedRepo"))?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
let removed = runner.prune_stale_workspaces()?;
assert_eq!(removed, 1, "only the orphan should be pruned");
assert!(!workspaces.join("999999").exists());
assert!(workspaces.join(recent.to_string()).exists());
assert!(workspaces.join("SharedRepo").exists());
Ok(())
}
// ── queue ────────────────────────────────────────────────────────
#[test]
fn queue_validates_refs() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
assert!(
runner
.queue("app", Some("--upload-pack=x"), None, "manual")
.is_err()
);
Ok(())
}
#[test]
fn queue_accepts_valid_ref() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
let queued = runner.queue("app", Some("refs/heads/feature"), None, "manual")?;
assert!(queued.id > 0);
Ok(())
}
#[test]
fn queue_uses_default_branch() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
let queued = runner.queue("app", None, None, "manual")?;
assert!(queued.id > 0);
Ok(())
}
#[test]
fn queue_deduplicates_active_commit() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
let first = runner.queue(
"app",
None,
Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
"manual",
)?;
let second = runner.queue(
"app",
None,
Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
"webhook",
)?;
assert_eq!(first.id, second.id);
assert!(!first.deduplicated);
assert!(second.deduplicated);
Ok(())
}
#[test]
fn queue_deduplicates_unresolved_webhook_deliveries() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
let first = runner.queue("app", None, None, "webhook")?;
let second = runner.queue("app", None, None, "webhook")?;
assert_eq!(first.id, second.id);
assert!(!first.deduplicated);
assert!(second.deduplicated);
Ok(())
}
#[test]
fn queue_supersedes_older_waiting_runs_on_same_ref() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database.clone(), root.path().to_owned(), false)?;
let stale = runner.queue(
"app",
None,
Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
"webhook",
)?;
database.finish_run(stale.id, "waiting", None)?;
let fresh = runner.queue(
"app",
None,
Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
"webhook",
)?;
assert!(!fresh.deduplicated);
assert_eq!(database.run(stale.id)?.status, "canceled");
assert_eq!(database.run(fresh.id)?.status, "queued");
Ok(())
}
#[test]
fn queue_rejects_invalid_commit() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
assert!(
runner
.queue("app", None, Some("not-a-sha"), "manual")
.is_err()
);
Ok(())
}
#[test]
fn queue_rejects_unknown_repository() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
assert!(runner.queue("nonexistent", None, None, "manual").is_err());
Ok(())
}
// ── environment isolation helpers ────────────────────────────────
#[test]
fn copy_host_environment_sets_well_known_vars() {
let mut cmd = Command::new("true");
copy_host_environment(&mut cmd);
// Just verify it doesn't panic — env vars may or may not exist
}
#[test]
#[cfg(unix)]
fn native_jobs_see_the_build_context() -> Result<()> {
let root = tempfile::tempdir()?;
let workspace = root.path().join("workspaces").join("7");
fs::create_dir_all(&workspace)?;
let runner = Runner::new(Database::memory(KEY)?, root.path().to_owned(), true)?;
let spec = JobSpec {
base_name: "context".into(),
name: "context".into(),
needs: Vec::new(),
executor: "native".into(),
image: None,
shell: None,
command: "echo \"root=$AKURAI_BUILD_ROOT\"; echo \"hosted=$AKURAI_BUILD_HOSTED\"; \
echo \"workspace=$AKURAI_BUILD_WORKSPACE\"; echo \"home=$AKURAI_BUILD_HOME\""
.into(),
matrix: BTreeMap::new(),
artifacts: Vec::new(),
cache: Vec::new(),
network: false,
secrets: Vec::new(),
installation_id: None,
installation_revision: None,
rollback_of_deployment_id: None,
recovery_of_deployment_id: None,
environment: None,
approval: false,
branches: Vec::new(),
timeout: 60,
profile: None,
};
let cancellation = RunCancellation::new();
let result = runner.execute(&spec, &workspace, 7, 1, &cancellation)?;
assert!(result.status.success(), "job failed: {}", result.logs);
assert!(
result
.logs
.contains(&format!("root={}", root.path().display())),
"{}",
result.logs
);
assert!(
result
.logs
.contains(&format!("hosted={}/hosted", root.path().display())),
"{}",
result.logs
);
assert!(
result
.logs
.contains(&format!("workspace={}", workspace.display())),
"{}",
result.logs
);
// HOME inside the job is the sandbox; AKURAI_BUILD_HOME is the real one.
let home = env::var("HOME")?;
assert!(
result.logs.contains(&format!("home={home}")),
"{}",
result.logs
);
assert_ne!(home, workspace.join(".home").display().to_string());
Ok(())
}
fn make_job_spec(base_name: &str, executor: &str, image: Option<&str>) -> JobSpec {
JobSpec {
base_name: base_name.into(),
name: base_name.into(),
needs: Vec::new(),
executor: executor.into(),
image: image.map(str::to_owned),
shell: None,
command: "true".into(),
matrix: BTreeMap::new(),
artifacts: Vec::new(),
cache: Vec::new(),
network: false,
secrets: Vec::new(),
installation_id: None,
installation_revision: None,
rollback_of_deployment_id: None,
recovery_of_deployment_id: None,
environment: None,
approval: false,
branches: Vec::new(),
timeout: 60,
profile: None,
}
}
#[test]
fn native_job_rejected_on_non_default_branch_of_trusted_repository() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let repo = database.add_repository("app", "https://example.com/app.git", "main")?;
assert!(repo.trusted);
let run_id = database.create_run(repo.id, "refs/heads/feature", None, "manual")?;
database.insert_jobs(run_id, &[make_job_spec("build", "native", None)])?;
let runner = Runner::new(database, root.path().to_owned(), true)?;
let workspace = root.path().join("workspaces").join(run_id.to_string());
fs::create_dir_all(&workspace)?;
let job = runner.database.jobs(run_id)?[0].id;
runner.execute_job(run_id, job, &workspace, &RunCancellation::new())?;
let job = &runner.database.jobs(run_id)?[0];
assert_eq!(job.status, "failed");
assert!(job.logs.contains("default branch"));
Ok(())
}
#[test]
fn native_job_allowed_on_default_branch_of_trusted_repository() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let repo = database.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = database.create_run(repo.id, "refs/heads/main", None, "manual")?;
database.insert_jobs(run_id, &[make_job_spec("build", "native", None)])?;
let runner = Runner::new(database, root.path().to_owned(), true)?;
let workspace = root.path().join("workspaces").join(run_id.to_string());
fs::create_dir_all(&workspace)?;
let job = runner.database.jobs(run_id)?[0].id;
runner.execute_job(run_id, job, &workspace, &RunCancellation::new())?;
let job = &runner.database.jobs(run_id)?[0];
assert_eq!(job.status, "succeeded", "{:?}", job.logs);
Ok(())
}
#[test]
fn untrusted_refs_reject_shared_caches_and_deployments() -> Result<()> {
for field in ["cache", "environment"] {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let repo = database.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = database.create_run(repo.id, "refs/heads/feature", None, "manual")?;
let image = format!("example@sha256:{}", "a".repeat(64));
let mut spec = make_job_spec("check", "docker", Some(&image));
match field {
"cache" => spec.cache.push("target".into()),
_ => spec.environment = Some("staging".into()),
}
database.insert_jobs(run_id, &[spec])?;
let runner = Runner::new(database, root.path().to_owned(), true)?;
let job = runner.database.jobs(run_id)?[0].id;
runner.execute_job(run_id, job, root.path(), &RunCancellation::new())?;
let job = &runner.database.jobs(run_id)?[0];
assert_eq!(job.status, "failed");
assert!(job.logs.contains("untrusted refs cannot"), "{}", job.logs);
}
Ok(())
}
#[test]
fn docker_job_on_pr_branch_of_trusted_repo_requires_immutable_image() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
let repo = database.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = database.create_run(repo.id, "refs/heads/feature", None, "manual")?;
// Mutable tag on a PR branch: rejected even though the repository is trusted.
database.insert_jobs(
run_id,
&[make_job_spec("verify", "docker", Some("alpine:3.22"))],
)?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
let workspace = root.path().join("workspaces").join(run_id.to_string());
fs::create_dir_all(&workspace)?;
let job = runner.database.jobs(run_id)?[0].id;
runner.execute_job(run_id, job, &workspace, &RunCancellation::new())?;
let job = &runner.database.jobs(run_id)?[0];
assert_eq!(job.status, "failed");
assert!(job.logs.contains("immutable sha256"));
Ok(())
}
#[test]
fn declared_secrets_requires_env_vars() {
// Without setting AKURAI_SECRET_FOO, this should fail
let names = vec!["NONEXISTENT_SECRET_XYZ".to_string()];
assert!(declared_secrets(&names).is_err());
}
}