Menu
AkurAI-Build
publicLatest change c992de8030667dbd1a2b0d9837220d74ba41b538 - Merge fc/report-docker-timeout-cleanup-failure by AkurAI Build
use std::{
collections::{BTreeMap, BTreeSet},
env,
fs::{self, File, OpenOptions},
io::{Read, Write},
path::{Path, PathBuf},
process::{Command, ExitStatus, Stdio},
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;
#[derive(Clone)]
pub struct Runner {
database: Database,
root: PathBuf,
allow_native: bool,
}
#[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,
}
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,
})
}
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"
);
if let Err(error) = self.execute_claimed(run_id) {
let message = bounded(&format!("{error:#}"), 16 * 1024);
self.database.finish_run(run_id, "failed", Some(&message))?;
}
self.database.detail(run_id)
}
fn execute_claimed(&self, run_id: i64) -> 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 {
let jobs = self.database.jobs(run_id)?;
if jobs.iter().all(|job| {
matches!(
job.status.as_str(),
"succeeded" | "failed" | "skipped" | "canceled" | "interrupted"
)
}) {
let failed = jobs.iter().any(|job| {
matches!(job.status.as_str(), "failed" | "canceled" | "interrupted")
});
self.database.finish_run(
run_id,
if failed { "failed" } else { "succeeded" },
None,
)?;
if !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 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)?;
progressed = true;
break;
}
if !progressed {
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) -> 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(());
}
if spec.executor == "native" && !self.allow_native {
self.database.set_job_status(job_id, "failed", Some("native execution is disabled; use a trusted repository and AKURAI_ALLOW_NATIVE=1"), 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);
match execution {
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) => {
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,
) -> 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 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);
}
run_process(command, spec.timeout, &secrets)
} else {
self.execute_docker(spec, workspace, run_id, job_id, &matrix, &secrets)
}
}
fn execute_docker(
&self,
spec: &JobSpec,
workspace: &Path,
run_id: i64,
job_id: i64,
matrix: &BTreeMap<String, String>,
secrets: &BTreeMap<String, String>,
) -> Result<ProcessResult> {
let name = format!("akurai-{run_id}-{job_id}");
let mut command = Command::new("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 = self.database.run(run_id)?.repository;
for path in &spec.cache {
let host = self
.root
.join("cache")
.join(safe_component(&repository))
.join(cache_component(path));
fs::create_dir_all(&host)?;
fs::create_dir_all(workspace.join(path))?;
command.args(["--volume", &format!("{}:/workspace/{path}", host.display())]);
}
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)?;
if result.timed_out {
match Command::new("docker")
.args(["rm", "--force", &name])
.output()
{
Ok(output) if !output.status.success() => result.logs.push_str(&format!(
"\nfailed to remove timed-out container {name}: {}",
String::from_utf8_lossy(&output.stderr)
)),
Err(error) => result.logs.push_str(&format!(
"\nfailed to remove timed-out container {name}: {error}"
)),
Ok(_) => {}
}
}
Ok(result)
}
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 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 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 = Command::new("/bin/sh");
command.args(["-eu", "-c", script]);
command
}
}
fn run_process(
mut command: Command,
timeout_seconds: u64,
secrets: &BTreeMap<String, String>,
) -> Result<ProcessResult> {
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = command.spawn().context("start build process")?;
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 (status, timed_out) = loop {
if let Some(status) = child.try_wait()? {
break (status, false);
}
if Instant::now() >= deadline {
child.kill()?;
break (child.wait()?, true);
}
thread::sleep(Duration::from_millis(100));
};
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,
})
}
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 safe_component(value: &str) -> String {
value
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() || matches!(character, '.' | '-' | '_') {
character
} else {
'_'
}
})
.collect()
}
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";
// ── safe_component ───────────────────────────────────────────────
#[test]
fn safe_component_preserves_safe_chars() {
assert_eq!(safe_component("hello"), "hello");
assert_eq!(safe_component("my-app"), "my-app");
assert_eq!(safe_component("v1.2.3"), "v1.2.3");
assert_eq!(safe_component("a_b"), "a_b");
}
#[test]
fn safe_component_replaces_unsafe_chars() {
assert_eq!(safe_component("a/b"), "a_b");
assert_eq!(safe_component("a b"), "a_b");
assert_eq!(safe_component("a;b"), "a_b");
assert_eq!(safe_component("../etc"), ".._etc");
assert_eq!(safe_component("$(cmd)"), "__cmd_");
}
#[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"));
}
// ── 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_sh_on_unix() {
let cmd = native_shell(None, "echo hello");
let program = cmd.get_program().to_string_lossy().into_owned();
#[cfg(not(windows))]
assert!(
program.contains("sh") || program.contains("bash"),
"expected sh-based shell, 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(&"echo hello".to_string()),
"args should contain script, got {args:?}"
);
}
// ── 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 result = runner.execute(&spec, &workspace, 7, 1)?;
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(())
}
#[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());
}
}