//! `sylphx build run [PATH] -- <command…>`: runs a command on a remote copy
//! of this git work tree and behaves like the local command (the remote
//! build execution ADR, section 1). The output streams back, the exit code is
//! the command's own, and named outputs are copied back.
//!
//! Every run leases a `build-<size>` machine with a cold workspace at [`WS`].
//! The workspace is created only after admission, belongs to the lease's
//! workload, and is reclaimed when the lease ends. Scoped remote build caches
//! reuse build outputs without retaining a local workspace.
//! Every run releases the lease. Before the command starts, a preempted or lost
//! machine, or a broken stream, is retried once after a short backoff, and
//! anything that fails (the machine, the sync, the toolchain install) is a
//! platform failure, 125. Once the command has started it is never run
//! again: its own status is the exit code, a broken output stream is
//! reattached to the running command (`Guest::reattach`), and a machine lost
//! under it ends the run with 137, as a killed process.
//!
//! `--region` pins placement to that Cell; otherwise the API selects Build
//! capacity. Neither path creates or reserves a persistent workspace Volume.
//! Once a lease is accepted, every internal retry stays in its returned Cell,
//! including a lease lost before READY.
//! `--queue-timeout` bounds the wait for a machine: past it the run answers
//! 125, retryable, so a caller can try another region.
//!
//! Exit codes: the command's own; 2 usage; 124 `--timeout`; 125 a platform
//! failure before the command started (with `retryable` in the `result`
//! event); 130 interrupted; 137 the machine was lost after the command
//! started.
mod git_deps;
mod history;
mod image;
mod remote;
mod store;
mod sync;
use std::collections::BTreeMap;
use std::io::Write;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;
use std::time::{Duration, Instant};
use clap::{Arg, ArgAction, ArgMatches, Command};
use serde_json::{json, Value};
use sylphx::sandboxes as sbx;
use sylphx::Client;
use crate::build_cache;
use crate::devices::{duration, parent_env, state, why};
use crate::Failure;
use sylphx_build_lease::guest::{Event, Fault, Guest, Reattach, REATTACH_BUDGET};
use sylphx_build_lease::scripts::{provision_env, ARTIFACTS, BOOTSTRAP, PROVISION, RUN};
use sylphx_build_lease::{
allowed_domains, judge, lease_request, provision_failure, wait_ready, LeaseParams, Ready,
Verdict, BUILD_CACHE_HOST, WS,
};
/// `sylphx build image`'s arguments.
pub fn image_command() -> Command {
image::command()
}
pub const EXIT_TIMEOUT: u8 = 124;
pub const EXIT_PLATFORM: u8 = 125;
pub const EXIT_INTERRUPTED: u8 = 130;
/// The command started and its machine was lost: 128 + SIGKILL, as a shell
/// reports a process the kernel killed, never 125 (that is only for anything
/// before the command starts).
pub const EXIT_LOST: u8 = 137;
const USER: &str = "user";
const SIZES: [&str; 3] = ["standard", "large", "xlarge"];
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60 * 60);
const MAX_TIMEOUT: Duration = Duration::from_secs(6 * 60 * 60);
/// The machine wait includes admission and a cold boot; retain 30m unless
/// the caller overrides it.
const DEFAULT_QUEUE_TIMEOUT: Duration = Duration::from_secs(30 * 60);
/// A data-plane call this often keeps a silent command from idling out.
const KEEPALIVE: Duration = Duration::from_secs(60);
/// Renew the lease token well before it expires.
const TOKEN_RENEW: Duration = Duration::from_secs(45 * 60);
/// The lease's egress allow-list: for an image build [`image::allowed_domains`]
/// (no internet package host), otherwise [`allowed_domains`] (the
/// `build-packages` preset, then each `--allow-host`).
fn lease_domains(o: &Opts) -> Vec<String> {
match &o.image {
Some(img) => image::allowed_domains(img, &o.allow_hosts),
None => allowed_domains(&o.allow_hosts),
}
}
pub fn command() -> Command {
Command::new("run")
.about("Run a command on a remote build machine against this work tree; output, exit code and --artifact files come back")
.long_about("Runs <command> on a remote copy of the enclosing git work tree, in PATH inside it. Every run gets a cold, lease-owned workspace created only after admission and reclaimed at lease end. Scoped remote caches reuse build outputs; no persistent workspace Volume is created or attached, including with --region. Output streams back unchanged apart from at most three `sylphx:` lines on stderr.\n\nThe run gets the project's scoped shared build cache (sccache, Turbo) through its environment; if the cache cannot be reached the run builds without it and says so on one `sylphx: warning:` line. --no-cache skips it. Cargo uses CARGO_INCREMENTAL=0 with sccache and CARGO_PROFILE_DEV_DEBUG=0. --env overrides these defaults. --fresh keeps the empty-tree intent; each lease's workspace is already fresh.\n\nExit codes: the command's own; 2 usage; 124 --timeout reached; 125 platform failure (not signed in, no capacity, no machine within --queue-timeout, region not offered, sync failed, the toolchain could not be installed, machine lost; anything before the command starts); 130 interrupted; 137 the machine was lost after the command started (it is never run twice; a broken output stream is reattached to the running command).\n\nKeep the default 30m machine wait for admission and cold boot; an explicit --queue-timeout replaces that budget.\n\nExample: sylphx build run --size standard -- cargo test -p sylphx-cli")
.arg(Arg::new("path").value_name("PATH").index(1)
.help("Directory to run in (default \".\"); the synced root is its git work tree"))
.arg(Arg::new("command").value_name("COMMAND").index(2).num_args(1..).last(true).required(true)
.help("The command and its arguments, after --"))
.arg(Arg::new("size").long("size").value_name("SIZE").default_value("large")
.value_parser(SIZES)
.help("Machine size: standard (8 vCPU), large (16), xlarge (32)"))
.arg(Arg::new("timeout").long("timeout").value_name("DURATION")
.help("Wall-clock limit for the command (default 60m, at most 6h)"))
.arg(Arg::new("region").long("region").value_name("REGION")
.help("Region to pin the run to (default: automatic Build placement); workspace lifetime always follows the lease"))
.arg(Arg::new("queue-timeout").long("queue-timeout").value_name("DURATION")
.help("Longest wait for a build machine before exiting 125, retryable (default 30m, at most 6h)"))
.arg(Arg::new("artifact").long("artifact").value_name("GLOB").action(ArgAction::Append)
.help("Copy matching files back after the run (repeatable; relative to PATH on the remote side)"))
.arg(Arg::new("out").long("out").value_name("DIR")
.help("Where artifacts land (default .sylphx/out/<run id>/)"))
.arg(Arg::new("env").long("env").value_name("NAME=VALUE").action(ArgAction::Append)
.help("Non-secret environment for the command (repeatable)"))
.arg(Arg::new("no-cache").long("no-cache").action(ArgAction::SetTrue)
.help("Do not use the shared build cache (no token is minted)"))
.arg(Arg::new("allow-host").long("allow-host").value_name("HOST").action(ArgAction::Append)
.help("Internet host the run may reach (repeatable); without it only the in-cell build cache and its crates and npm mirrors"))
.arg(Arg::new("source-history").long("source-history").value_name("REF")
.help("Send HEAD and REF's full Git history offline (e.g. origin/main); keep synced local changes"))
.arg(Arg::new("fresh").long("fresh").action(ArgAction::SetTrue)
.help("Start from an empty workspace: no synced tree, no target/ or sccache"))
.arg(Arg::new("dry-run").long("dry-run").action(ArgAction::SetTrue)
.help("Print what a cold run would send (files, bytes) and exit 0"))
.arg(Arg::new("quiet").long("quiet").short('q').action(ArgAction::SetTrue)
.help("No `sylphx:` progress lines on stderr"))
.arg(Arg::new("no-tree").long("no-tree").action(ArgAction::SetTrue)
.conflicts_with_all(["path", "artifact", "out", "env", "region", "fresh", "no-cache", "dry-run", "source-history"])
.help("Run on the Build service with no input tree: the run is a Build record, its output streams back from its log"))
.arg(Arg::new("no-wait").long("no-wait").action(ArgAction::SetTrue).requires("no-tree")
.help("Create the Build and print it without waiting; read the output later with `sylphx build logs`"))
}
/// `sylphx build logs <build> [--follow]`: the log of a Build on the service.
pub fn logs_command() -> Command {
Command::new("logs")
.about("Print a build's log; with --follow, keep printing until the build ends")
.long_about("Prints the log of a Build created by `sylphx build run --no-tree`, oldest line first: the command's stdout to stdout, its stderr and the platform's `sylphx:` lines to stderr. <BUILD> is a build name (orgs/{org}/projects/{project}/builds/{build}) or the id of a build in the linked project. With --follow the command waits for new lines and returns when the build has ended; without it, it prints what is written now.\n\nExample: sylphx build logs b01j8x --follow")
.arg(Arg::new("build").value_name("BUILD").required(true).index(1)
.help("The build name or id"))
.arg(Arg::new("follow").long("follow").short('f').action(ArgAction::SetTrue)
.help("Wait for new lines until the build ends"))
}
/// `sylphx build logs`.
pub async fn logs(client: Result<Client, Failure>, m: &ArgMatches) -> Result<(), Failure> {
let client = client?;
let arg = m.get_one::<String>("build").cloned().unwrap_or_default();
let name = if arg.starts_with("orgs/") {
arg
} else {
let project = match remote::project(
&client,
m.try_get_one::<String>("org")
.ok()
.flatten()
.map(String::as_str),
)
.await
{
Ok(p) => p,
Err(Outcome::Platform { reason, .. }) => return Err(Failure::Usage(reason)),
Err(_) => return Err(Failure::Exit(EXIT_PLATFORM)),
};
format!("{project}/builds/{arg}")
};
let out = Out {
json: false,
quiet: false,
};
tokio::select! {
r = remote::follow(&client, &name, &out, m.get_flag("follow")) => match r {
Ok(()) => Ok(()),
Err(Outcome::Platform { reason, .. }) => Err(Failure::Refused(reason)),
Err(_) => Err(Failure::Exit(EXIT_PLATFORM)),
},
_ = interrupted() => Err(Failure::Exit(EXIT_INTERRUPTED)),
}
}
#[derive(Debug)]
pub struct Opts {
root: PathBuf,
rel: String,
command: Vec<String>,
size: String,
timeout: Duration,
region: Option<String>,
queue_timeout: Duration,
artifacts: Vec<String>,
out: Option<PathBuf>,
env: BTreeMap<String, String>,
org: Option<String>,
allow_hosts: Vec<String>,
fresh: bool,
source_history: Option<String>,
no_cache: bool,
dry_run: bool,
quiet: bool,
json: bool,
/// Set for `build image`: `command` is rendered from it.
image: Option<image::Image>,
/// Run as a Build on the service, over no tree.
no_tree: bool,
no_wait: bool,
}
/// The limits and placement every kind of build takes alike.
struct Shared {
root: PathBuf,
rel: String,
size: String,
timeout: Duration,
region: Option<String>,
queue_timeout: Duration,
allow_hosts: Vec<String>,
}
impl Shared {
/// `no_tree`: a run over no tree has no work tree to find.
fn parse(m: &ArgMatches, no_tree: bool) -> Result<Self, String> {
let dir = PathBuf::from(
m.get_one::<String>("path")
.map(String::as_str)
.unwrap_or("."),
);
if !no_tree && !dir.is_dir() {
return Err(format!("{} is not a directory", dir.display()));
}
let (root, rel) = if no_tree {
(PathBuf::new(), String::new())
} else {
sync::work_tree(&dir)?
};
let timeout = match m.get_one::<String>("timeout") {
Some(v) => duration(v, "--timeout")?,
None => DEFAULT_TIMEOUT,
};
if timeout.is_zero() || timeout > MAX_TIMEOUT {
return Err("--timeout is more than 0s and at most 6h".into());
}
let queue_timeout = match m.get_one::<String>("queue-timeout") {
Some(v) => duration(v, "--queue-timeout")?,
None => DEFAULT_QUEUE_TIMEOUT,
};
if queue_timeout.is_zero() || queue_timeout > MAX_TIMEOUT {
return Err("--queue-timeout is more than 0s and at most 6h".into());
}
let region = m.get_one::<String>("region").cloned();
if let Some(r) = ®ion {
let ok = !r.is_empty()
&& r.len() <= 32
&& r.starts_with(|c: char| c.is_ascii_lowercase())
&& r.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
if !ok {
return Err(format!(
"--region {r}: give a region name such as gra (lower-case letters, digits, -)"
));
}
}
let mut allow_hosts = Vec::new();
for h in m.get_many::<String>("allow-host").into_iter().flatten() {
let ok = !h.is_empty()
&& h.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '*');
if !ok {
return Err(format!(
"--allow-host {h}: give a host name such as api.example.com"
));
}
allow_hosts.push(h.to_ascii_lowercase());
}
Ok(Self {
root,
rel,
size: m
.get_one::<String>("size")
.cloned()
.unwrap_or_else(|| "large".into()),
timeout,
region,
queue_timeout,
allow_hosts,
})
}
}
impl Opts {
/// `sylphx build run`.
pub fn parse(m: &ArgMatches) -> Result<Self, String> {
let no_tree = m.get_flag("no-tree");
let sh = Shared::parse(m, no_tree)?;
let command: Vec<String> = m
.get_many::<String>("command")
.map(|v| v.cloned().collect())
.unwrap_or_default();
if command.is_empty() || command[0].is_empty() {
return Err("give the command after --: sylphx build run -- cargo test".into());
}
let mut env = BTreeMap::new();
for kv in m.get_many::<String>("env").into_iter().flatten() {
let (k, v) = kv
.split_once('=')
.ok_or_else(|| format!("--env {kv}: give NAME=VALUE"))?;
let ok = !k.is_empty()
&& !k.starts_with(|c: char| c.is_ascii_digit())
&& k.chars().all(|c| c.is_ascii_alphanumeric() || c == '_');
if !ok {
return Err(format!("--env {kv}: `{k}` is not a variable name"));
}
if k == crate::context::ENV_VAR {
return Err(format!(
"--env {k}: it names this machine's default environment and is never sent to the run"
));
}
env.insert(k.to_string(), v.to_string());
}
Ok(Self {
org: m.try_get_one::<String>("org").ok().flatten().cloned(),
root: sh.root,
rel: sh.rel,
command,
size: sh.size,
timeout: sh.timeout,
region: sh.region,
queue_timeout: sh.queue_timeout,
artifacts: m
.get_many::<String>("artifact")
.map(|v| v.cloned().collect())
.unwrap_or_default(),
out: m.get_one::<String>("out").map(PathBuf::from),
env,
allow_hosts: sh.allow_hosts,
fresh: m.get_flag("fresh"),
source_history: m.get_one::<String>("source-history").cloned(),
no_cache: m.get_flag("no-cache"),
dry_run: m.get_flag("dry-run"),
quiet: m.get_flag("quiet"),
json: m.get_one::<String>("output").map(String::as_str) == Some("json"),
image: None,
no_tree,
no_wait: m.get_flag("no-wait"),
})
}
/// `sylphx build image`: the command is `sylphx-image-build` on the tree.
pub fn parse_image(m: &ArgMatches) -> Result<Self, String> {
let sh = Shared::parse(m, false)?;
let image = image::Image::parse(m)?;
Ok(Self {
org: m.try_get_one::<String>("org").ok().flatten().cloned(),
command: image.argv(&sh.rel),
root: sh.root,
rel: sh.rel,
size: sh.size,
timeout: sh.timeout,
region: sh.region,
queue_timeout: sh.queue_timeout,
artifacts: Vec::new(),
out: None,
env: BTreeMap::new(),
allow_hosts: sh.allow_hosts,
fresh: false,
source_history: None,
no_cache: false,
dry_run: false,
quiet: m.get_flag("quiet"),
json: m.get_one::<String>("output").map(String::as_str) == Some("json"),
image: Some(image),
no_tree: false,
no_wait: false,
})
}
}
/// How a run ended.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
/// The command ran to its end with this exit status.
Ran(i32),
TimedOut,
Interrupted,
/// The platform could not run it; `retryable` says whether trying again
/// may succeed.
Platform {
reason: String,
retryable: bool,
},
/// The command started and the run lost it (the machine ended under
/// it): it is not run again, and the run exits as a killed process does.
Lost(String),
}
impl Outcome {
pub fn code(&self) -> u8 {
match self {
// A local shell reports a status the same way: the low byte.
Outcome::Ran(c) => (*c & 0xff) as u8,
Outcome::TimedOut => EXIT_TIMEOUT,
Outcome::Interrupted => EXIT_INTERRUPTED,
Outcome::Platform { .. } => EXIT_PLATFORM,
Outcome::Lost(_) => EXIT_LOST,
}
}
fn name(&self) -> &'static str {
match self {
Outcome::Ran(0) => "succeeded",
Outcome::Ran(_) => "failed",
Outcome::TimedOut => "timed_out",
Outcome::Interrupted => "interrupted",
Outcome::Platform { .. } => "platform_error",
Outcome::Lost(_) => "lost",
}
}
fn platform(reason: impl Into<String>, retryable: bool) -> Self {
Outcome::Platform {
reason: reason.into(),
retryable,
}
}
}
/// Where progress, output and events go: raw output and `sylphx:` lines, or
/// NDJSON events on stdout with `-o json`.
struct Out {
json: bool,
quiet: bool,
}
impl Out {
fn progress(&self, text: &str) {
if !self.json && !self.quiet {
eprintln!("sylphx: {text}");
}
}
fn event(&self, v: Value) {
if self.json {
let mut o = std::io::stdout().lock();
let _ = writeln!(o, "{v}");
let _ = o.flush();
}
}
fn stdout(&self, b: &[u8]) {
if self.json {
self.event(json!({"type": "stdout", "data": String::from_utf8_lossy(b)}));
} else {
let mut o = std::io::stdout().lock();
let _ = o.write_all(b);
let _ = o.flush();
}
}
fn stderr(&self, b: &[u8]) {
if self.json {
self.event(json!({"type": "stderr", "data": String::from_utf8_lossy(b)}));
} else {
let mut e = std::io::stderr().lock();
let _ = e.write_all(b);
let _ = e.flush();
}
}
}
/// What the way out needs even when the run itself was dropped (Ctrl-C).
#[derive(Default)]
struct Held(Mutex<HeldState>);
#[derive(Default)]
struct HeldState {
lease: Option<String>,
proc: Option<(Guest, u64)>,
/// Cache token values: removed from any text that leaves the process.
secrets: Vec<String>,
}
impl Held {
fn with<R>(&self, f: impl FnOnce(&mut HeldState) -> R) -> R {
f(&mut self.0.lock().expect("not poisoned"))
}
}
/// The counters of the `result` event.
#[derive(Default)]
struct Stats {
bytes_up: u64,
bytes_down: u64,
artifacts: usize,
/// Files the build machine took from the build store, not the upload.
from_store: usize,
/// `build image`: the script's metadata object.
image: Option<Value>,
}
/// `sylphx build run`. `client` is the signed-in client, or why there is none.
pub async fn run(
client: Result<Client, Failure>,
key: Option<String>,
m: &ArgMatches,
) -> Result<(), Failure> {
let opts = Opts::parse(m).map_err(Failure::Usage)?;
run_opts(client, key, opts).await
}
/// `sylphx build image`: the same machinery, with `sylphx-image-build` as the
/// command ([`image`]).
pub async fn run_image(
client: Result<Client, Failure>,
key: Option<String>,
m: &ArgMatches,
) -> Result<(), Failure> {
let mut opts = Opts::parse_image(m).map_err(Failure::Usage)?;
if let Some(img) = opts.image.as_mut() {
img.source = image::source_of(&opts.root);
}
run_opts(client, key, opts).await
}
async fn run_opts(
client: Result<Client, Failure>,
key: Option<String>,
opts: Opts,
) -> Result<(), Failure> {
let out = Out {
json: opts.json,
quiet: opts.quiet,
};
let t0 = Instant::now();
if opts.image.as_ref().is_some_and(|i| i.source.is_none()) {
out.progress("the work tree has local changes or no commit: building without SOURCE_DATE_EPOCH and a source commit, so the image is not reproducible");
}
if opts.dry_run {
return dry_run(&opts, &out);
}
let client = match client {
Ok(c) => c,
Err(f) => {
let reason = match f {
Failure::Usage(s) | Failure::Refused(s) => s,
Failure::Api(e) => why(&e),
Failure::Exit(_) => "not signed in".into(),
};
return finish(
&out,
&Outcome::platform(reason, false),
&Stats::default(),
t0,
);
}
};
if opts.no_tree {
let created = Mutex::new(None::<String>);
let ended = tokio::select! {
o = remote::run(&client, &opts, &out, &created) => o,
_ = interrupted() => remote::Ended::Outcome(Outcome::Interrupted),
};
// The Build outlives this process unless it is stopped.
if matches!(ended, remote::Ended::Outcome(Outcome::Interrupted)) {
let name = created.lock().ok().and_then(|mut c| c.take());
if let Some(name) = name {
remote::cancel(&client, &name).await;
}
}
return match ended {
// `--no-wait` printed the Build and is done.
remote::Ended::Created => Ok(()),
remote::Ended::Outcome(o) => finish(&out, &o, &Stats::default(), t0),
};
}
let held = Held::default();
let stats = Mutex::new(Stats::default());
let outcome = tokio::select! {
o = execute(&client, key.as_deref(), &opts, &out, &held, &stats) => o,
_ = interrupted() => Outcome::Interrupted,
};
// Stop the command first, so nothing more is written to the workspace,
// then end the machine. Either is bounded: an unreachable API must not
// hold the caller.
let (lease, proc) = held.with(|h| (h.lease.take(), h.proc.take()));
let outcome = scrub(outcome, &held.with(|h| h.secrets.clone()));
if matches!(outcome, Outcome::Interrupted) {
if let Some((g, pid)) = proc {
let _ = tokio::time::timeout(Duration::from_secs(5), g.signal(pid, "SIGKILL")).await;
}
}
if let Some(name) = lease {
release(&client, &name).await;
}
let stats = stats.into_inner().unwrap_or_default();
finish(&out, &outcome, &stats, t0)
}
async fn interrupted() {
#[cfg(unix)]
{
use tokio::signal::unix::{signal, SignalKind};
if let Ok(mut term) = signal(SignalKind::terminate()) {
tokio::select! {
_ = tokio::signal::ctrl_c() => {}
_ = term.recv() => {}
}
return;
}
}
let _ = tokio::signal::ctrl_c().await;
}
async fn release(client: &Client, name: &str) {
if !sylphx_build_lease::release(client, name).await {
eprintln!("sylphx: warning: {name} was not released; its idle timeout ends it");
}
}
/// The final `-o json` event.
fn result_event(outcome: &Outcome, s: &Stats, ms: u64) -> Value {
let mut result = json!({
"type": "result",
"exit_code": outcome.code(),
"outcome": outcome.name(),
"retryable": matches!(outcome, Outcome::Platform { retryable: true, .. } | Outcome::Lost(_)),
"duration_ms": ms,
"workspace": "cold",
"bytes_up": s.bytes_up,
"bytes_down": s.bytes_down,
"from_store": s.from_store,
});
if let Outcome::Platform { reason, .. } | Outcome::Lost(reason) = outcome {
result["error"] = json!(reason);
}
if let Some(image) = &s.image {
result["image"] = image.clone();
}
result
}
/// The `result` event and summary line, and the exit status.
fn finish(out: &Out, outcome: &Outcome, s: &Stats, t0: Instant) -> Result<(), Failure> {
let ms = t0.elapsed().as_millis() as u64;
if let Outcome::Platform { reason, retryable } = outcome {
eprintln!(
"sylphx: error: {reason}{}",
if *retryable { " [retryable]" } else { "" }
);
}
if let Outcome::Lost(reason) = outcome {
eprintln!(
"sylphx: error: {reason} after the command started; it was not run again [retryable]"
);
}
out.event(result_event(outcome, s, ms));
let what = match outcome {
Outcome::Ran(c) => format!("exit {c}"),
Outcome::TimedOut => "timed out".into(),
Outcome::Interrupted => "interrupted".into(),
Outcome::Platform { .. } => "platform failure".into(),
Outcome::Lost(_) => "lost the machine".into(),
};
let arts = match s.artifacts {
0 => String::new(),
1 => "; 1 artifact".into(),
n => format!("; {n} artifacts"),
};
out.progress(&format!("{what} in {:.1} s{arts}", ms as f64 / 1000.0));
match outcome.code() {
0 => Ok(()),
c => Err(Failure::Exit(c)),
}
}
fn dry_run(o: &Opts, out: &Out) -> Result<(), Failure> {
let files = sync::file_set(&o.root).map_err(Failure::Usage)?;
let mut cache = StatCache::open(&o.root);
let tree = sync::scan(&o.root, &files, &mut cache.cache).map_err(Failure::Usage)?;
cache.save();
let plan = sync::plan(&tree, None);
let bytes = plan.upload_bytes(&tree);
out.event(json!({"type": "sync", "dry_run": true, "full": true, "files": plan.upload.len(), "bytes": bytes, "removed": 0}));
if !o.json {
println!(
"{} files, {} to send to a cold lease-owned workspace",
plan.upload.len(),
human(bytes)
);
}
Ok(())
}
struct StatCache {
path: Option<PathBuf>,
cache: sync::StatCache,
}
impl StatCache {
fn open(root: &Path) -> Self {
let path = sync::cache_path(root);
let cache = path
.as_deref()
.map(sync::StatCache::load)
.unwrap_or_default();
Self { path, cache }
}
fn save(&self) {
if let Some(p) = &self.path {
let _ = self.cache.save(p);
}
}
}
fn human(b: u64) -> String {
if b >= 1 << 20 {
format!("{:.1} MiB", b as f64 / 1048576.0)
} else if b >= 1 << 10 {
format!("{:.0} KiB", b as f64 / 1024.0)
} else {
format!("{b} B")
}
}
async fn execute(
client: &Client,
key: Option<&str>,
o: &Opts,
out: &Out,
held: &Held,
stats: &Mutex<Stats>,
) -> Outcome {
let parent = match parent_env(client, None, o.org.as_deref()).await {
Ok(p) => p,
Err(e) => return Outcome::platform(e, false),
};
// The tree is hashed once and reused on a fresh workspace after a retry.
let files = match sync::file_set(&o.root) {
Ok(f) => f,
Err(e) => return Outcome::platform(format!("listing the work tree: {e}"), false),
};
let mut cache = StatCache::open(&o.root);
let tree = match sync::scan(&o.root, &files, &mut cache.cache) {
Ok(t) => t,
Err(e) => return Outcome::platform(format!("hashing the work tree: {e}"), false),
};
cache.save();
// Git dependencies are fetched here, where the credentials are, before
// any machine is leased: one that cannot be fetched fails in seconds.
let git_deps = if o.image.is_some() {
Vec::new()
} else {
match prepare_git_deps(&o.root, &tree, &o.rel) {
Ok(d) => d,
Err(e) => return Outcome::platform(e, false),
}
};
// Bundling can take longer than a machine's idle window. Do it once,
// while interruptible and before any lease, then borrow it on each guest.
let source_history = match o.source_history.as_deref() {
Some(base) => match history::prepare(&o.root, base).await {
Ok(history) => Some(history),
Err(e) => return Outcome::platform(e, false),
},
None => None,
};
let mut run = Run {
client,
o,
out,
held,
stats,
parent,
tree,
git_deps,
source_history,
key,
cache_url: build_cache::base_url(),
cache_warned: AtomicBool::new(false),
accepted_region: None,
};
let mut retries = Retries::default();
loop {
match retries.next(run.attempt().await) {
Step::Done(outcome) => return outcome,
Step::Again { reason, backoff } => {
out.progress(&format!("{reason}; retrying once"));
out.event(json!({"type": "queued", "retry": true, "reason": reason}));
tokio::time::sleep(backoff).await;
}
}
}
}
/// Waited before the one retry, so a gateway that dropped a stream is not hit
/// again in the same instant.
const RETRY_BACKOFF: Duration = Duration::from_secs(5);
/// The one retry a run gets before its command starts, whatever lost it: a
/// machine preempted or lost, or a process stream that broke.
#[derive(Default)]
struct Retries {
used: bool,
}
enum Step {
Done(Outcome),
Again { reason: String, backoff: Duration },
}
impl Retries {
fn next(&mut self, a: Attempt) -> Step {
match a {
Attempt::Done(o) => Step::Done(o),
Attempt::Lost(reason) if !self.used => {
self.used = true;
Step::Again {
reason,
backoff: RETRY_BACKOFF,
}
}
Attempt::Lost(reason) => {
Step::Done(Outcome::platform(format!("{reason}, twice"), true))
}
}
}
}
/// Why [`Run::exec`] lost the command, and whether it had started: only a
/// command that never started may be run again.
struct ExecFault {
started: bool,
fault: Fault,
}
impl ExecFault {
fn before(fault: Fault) -> Self {
ExecFault {
started: false,
fault,
}
}
}
enum Attempt {
Done(Outcome),
/// The machine was preempted or lost: worth one more try.
Lost(String),
}
struct Run<'a> {
client: &'a Client,
o: &'a Opts,
out: &'a Out,
held: &'a Held,
stats: &'a Mutex<Stats>,
parent: String,
tree: sync::Tree,
/// The tree's locked git dependencies and their tarballs on this machine.
git_deps: Vec<(git_deps::Dep, PathBuf)>,
/// HEAD and the requested ref, bundled once before acquiring any lease.
source_history: Option<(String, Vec<u8>)>,
/// The caller's key, for minting the cache token.
key: Option<&'a str>,
cache_url: String,
/// One warning per run: the build cache unavailable, or the build store
/// not taking the sync.
cache_warned: AtomicBool,
/// The first accepted lease's canonical region, even before READY.
/// `None` is unselected; `Some("")` is the legacy home-Cell spelling.
accepted_region: Option<String>,
}
/// A ready machine with its workspace.
struct Machine {
lease: sbx::Lease,
}
impl Run<'_> {
/// The accepted Cell, otherwise the explicit region or automatic placement.
fn region(&self) -> &str {
match self.accepted_region.as_deref() {
// An empty CreateLease region would select automatically again.
Some("") => "hk",
Some(region) => region,
None => self.o.region.as_deref().unwrap_or(""),
}
}
/// " in region `<r>`" when a region was asked for.
fn where_(&self) -> String {
match &self.o.region {
Some(r) => format!(" in region {r}"),
None => String::new(),
}
}
fn add(&self, f: impl FnOnce(&mut Stats)) {
f(&mut self.stats.lock().expect("not poisoned"));
}
async fn attempt(&mut self) -> Attempt {
let m = match self.acquire().await {
Ok(m) => m,
Err(a) => return a,
};
let name = m.lease.name.clone();
let mut guest = match self.guest(&m.lease).await {
Ok(g) => g,
Err(e) => return self.lost_or(&name, e).await,
};
let mut minted = Instant::now();
let warned_before = self.cache_warned.load(Ordering::Relaxed);
// Minted before the sync: the build store takes the sync's files.
let (cache, store) = self.cache(&m.lease).await;
match self.sync(&guest, store.as_ref()).await {
Ok(()) => {}
Err(e) if workspace_full(e.text()) => return self.out_of_space(&m).await,
Err(e) => return self.lost_or(&name, e).await,
}
let git_env = match self.send_git_deps(&guest).await {
Ok(e) => e,
Err(e) if out_of_space(e.text()) => return self.out_of_space(&m).await,
Err(e) => return self.lost_or(&name, e).await,
};
// An image build needs no Rust toolchain.
if self.o.image.is_none() {
if let Err(a) = self.provision(&guest, &name, &cache).await {
if matches!(&a, Attempt::Done(Outcome::Platform { reason, .. }) if out_of_space(reason))
{
return self.out_of_space(&m).await;
}
return a;
}
}
if let Err(e) = self.send_source_history(&guest).await {
return self.lost_or(&name, e).await;
}
let shape = format!("build-{}", self.o.size);
let warned = !warned_before && self.cache_warned.load(Ordering::Relaxed);
// At most three `sylphx:` lines: a warning takes the place of this one.
if !warned {
self.out.progress(&format!(
"running on {shape}, cold lease-owned workspace{}",
self.where_()
));
}
self.out.event(
json!({"type": "running", "lease": name, "size": self.o.size, "region": m.lease.spec.as_ref().map(|s| s.region.as_str()).unwrap_or(self.region()), "workspace": "cold", "volume": false, "cache": !cache.is_empty()}),
);
if let Err(e) = self.send_auth(&guest).await {
return self.lost_or(&name, e).await;
}
let ran = self
.exec(&mut guest, &mut minted, &name, cache, git_env)
.await;
// The credentials are gone before anything else is asked of the
// machine, whatever the build's end.
self.drop_auth(&guest).await;
let code = match ran {
Ok(Some(c)) => c,
Ok(None) => return Attempt::Done(Outcome::TimedOut),
Err(ExecFault {
started: false,
fault,
}) => return self.lost_or(&name, fault).await,
Err(ExecFault {
started: true,
fault,
}) => return self.lost_after_start(&name, fault).await,
};
if self.o.image.is_some() && code == 0 {
if let Err(o) = self.collect_image(&guest, &name).await {
return Attempt::Done(o);
}
}
if !self.o.artifacts.is_empty() {
if let Err(e) = self.collect(&guest, &name).await {
return Attempt::Done(Outcome::platform(
format!("copying artifacts back: {e}"),
true,
));
}
}
Attempt::Done(Outcome::Ran(code))
}
/// The cache environment for the command, minted before the sync (its
/// lifetime is the run's timeout plus slack, which covers the sync). Empty
/// with `--no-cache`, and when no token
/// can be had (a cache that is missing only slows the build). The build
/// store, when the token reaches it, takes the sync's files.
async fn cache(&self, lease: &sbx::Lease) -> (BTreeMap<String, String>, Option<store::Store>) {
if self.o.no_cache {
return (BTreeMap::new(), None);
}
let minted = match (self.key, build_cache::project_id(&self.parent)) {
(Some(key), Some(project)) => {
let req = build_cache::run_request(project, self.o.timeout, lease);
build_cache::mint(&self.cache_url, key, &req).await
}
(None, _) => Err(build_cache::MintError::NoProject("not signed in".into())),
(_, None) => Err(build_cache::MintError::NoProject(
"the environment names no project".into(),
)),
};
match minted {
Ok(m) => {
self.held.with(|h| h.secrets.extend(m.secrets()));
// The gateway as the build machine reaches it.
let guest_base = m.env.get("TURBO_API").cloned().unwrap_or_default();
let st = m.token.as_deref().and_then(|t| {
store::Store::new(
&self.cache_url,
&guest_base,
m.cache.as_deref().unwrap_or(""),
t,
)
});
(m.env, st)
}
Err(e) => {
let first = !self.cache_warned.swap(true, Ordering::Relaxed);
if first {
eprintln!(
"sylphx: warning: build cache unavailable ({}); building without it",
e.short()
);
}
(BTreeMap::new(), None)
}
}
}
/// Installs what the command needs before it starts. Any failure here is
/// a platform failure (125, retryable), never the command's own status.
async fn provision(
&self,
g: &Guest,
name: &str,
cache: &BTreeMap<String, String>,
) -> Result<(), Attempt> {
let args = argv(&[
"/bin/sh",
"-c",
PROVISION,
"sylphx-provision",
WS,
self.rel(),
]);
let env = provision_env(cache);
let ran = tokio::time::timeout(PROVISION_TIMEOUT, g.run(USER, &args, &env)).await;
match ran {
Err(_) => Err(Attempt::Done(Outcome::platform(
format!(
"preparing the build machine took longer than {}s",
PROVISION_TIMEOUT.as_secs()
),
true,
))),
Ok(Ok((0, _, _))) => Ok(()),
Ok(Ok((code, _, err))) => Err(Attempt::Done(Outcome::platform(
provision_failure(code, &err),
true,
))),
Ok(Err(f)) => Err(self.lost_or(name, f).await),
}
}
/// The tree directory the command runs in, relative to the tree root.
fn rel(&self) -> &str {
if self.o.rel.is_empty() {
"."
} else {
&self.o.rel
}
}
/// A guest fault means the machine may be gone: if the lease ended
/// preempted or lost, or the process stream broke, the run is worth one
/// retry; otherwise it is a platform failure.
async fn lost_or(&self, name: &str, f: Fault) -> Attempt {
let mut get = sbx::GetLeaseRequest::default();
get.name = name.to_string();
let lease = self.client.sandboxes().leases().get(get).await.ok();
let reason = lease
.as_ref()
.and_then(|l| l.status.as_ref())
.and_then(|s| s.end_reason.as_ref())
.map(|r| r.as_str().to_string());
let ended = lease
.as_ref()
.map(|l| matches!(state(l), sbx::LeaseState::Ending | sbx::LeaseState::Ended))
.unwrap_or(false);
let verdict = judge(reason.as_deref(), &f);
// A retry takes a new machine: end this one first, or it idles on.
let release_now = !ended && matches!(verdict, Verdict::Retry(_));
self.held.with(|h| {
h.proc = None;
if ended || release_now {
h.lease = None;
}
});
if release_now {
release(self.client, name).await;
}
match verdict {
Verdict::Retry(why) => Attempt::Lost(why),
Verdict::Fail { reason, retryable } => {
Attempt::Done(Outcome::platform(reason, retryable))
}
}
}
/// Retry once before the command starts on another fresh lease-owned
/// workspace. A second failure is 125, as any loss before the start.
async fn out_of_space(&mut self, m: &Machine) -> Attempt {
let name = &m.lease.name;
self.held.with(|h| {
h.proc = None;
h.lease = None;
});
release(self.client, name).await;
Attempt::Lost("the workspace ran out of space".into())
}
/// The command started and the run lost it: the machine ended, or the
/// command cannot be followed. It is never run again (it may have done
/// part of its work); the run ends as a killed process would.
async fn lost_after_start(&self, name: &str, f: Fault) -> Attempt {
let mut get = sbx::GetLeaseRequest::default();
get.name = name.to_string();
let lease = self.client.sandboxes().leases().get(get).await.ok();
let reason = lease
.as_ref()
.and_then(|l| l.status.as_ref())
.and_then(|s| s.end_reason.as_ref())
.map(|r| format!("the build machine ended: {}", r.as_str()))
.unwrap_or_else(|| f.text().to_string());
self.held.with(|h| h.proc = None);
Attempt::Done(Outcome::Lost(reason))
}
/// Lease first; the workload creates its cold workspace only after
/// admission. Explicit regions pin placement, never a workspace Volume.
async fn acquire(&mut self) -> Result<Machine, Attempt> {
let deadline = Instant::now() + self.o.queue_timeout;
let lease = self.lease().await.map_err(Attempt::Done)?;
self.out.event(
json!({"type": "queued", "lease": lease.name, "size": self.o.size, "region": lease.spec.as_ref().map(|s| s.region.as_str()).unwrap_or(self.region()), "volume": false}),
);
self.ready(lease, deadline)
.await
.map(|lease| Machine { lease })
}
/// Create a Build lease with no pre-created or attached workspace Volume.
async fn lease(&mut self) -> Result<sbx::Lease, Outcome> {
let domains = lease_domains(self.o);
let req = lease_request(
&self.parent,
&LeaseParams {
size: &self.o.size,
region: self.region(),
timeout: self.o.timeout,
domains: &domains,
},
);
let lease = self
.client
.sandboxes()
.leases()
.create(req)
.await
.map_err(|e| api("creating the build machine", &e))?;
if self.accepted_region.is_none() {
self.accepted_region = lease.spec.as_ref().map(|s| s.region.clone());
}
self.held.with(|h| h.lease = Some(lease.name.clone()));
Ok(lease)
}
/// Wait for admission and READY within the caller's queue budget.
async fn ready(&self, lease: sbx::Lease, deadline: Instant) -> Result<sbx::Lease, Attempt> {
match wait_ready(self.client, lease, None, deadline).await {
Ready::Ready(l) => Ok(*l),
// No workspace node pin is requested by this caller.
Ready::PinExpired => unreachable!("volume-free Build leases have no node pin"),
Ready::QueueTimeout => {
self.held.with(|h| h.lease = None);
Err(Attempt::Done(Outcome::platform(
format!(
"no build machine{} within --queue-timeout {}s",
self.where_(),
self.o.queue_timeout.as_secs()
),
true,
)))
}
Ready::Refused { why, retryable } => {
self.held.with(|h| h.lease = None);
Err(Attempt::Done(Outcome::platform(
format!("the build machine was refused: {why}"),
retryable,
)))
}
Ready::Ended { reason, retryable } => {
self.held.with(|h| h.lease = None);
Err(Attempt::Done(Outcome::platform(
format!("the build machine ended before it was ready: {reason}"),
retryable,
)))
}
Ready::Lost(reason) => {
self.held.with(|h| h.lease = None);
Err(Attempt::Lost(reason))
}
Ready::Api(e) => Err(Attempt::Done(api("reading the build machine", &e))),
}
}
async fn token(&self, lease: &str) -> Result<String, Fault> {
sylphx_build_lease::mint_token(self.client, lease).await
}
async fn guest(&self, lease: &sbx::Lease) -> Result<Guest, Fault> {
sylphx_build_lease::guest(self.client, lease).await
}
/// Copy the current tree into a fresh workspace. Full/cold syncs retain
/// the direct compressed tarball path rather than one fetch per file.
async fn sync(&self, g: &Guest, store: Option<&store::Store>) -> Result<(), Fault> {
let t = Instant::now();
let none = BTreeMap::new();
let (code, _, err) = g
.run(
"root",
&argv(&["/bin/sh", "-c", BOOTSTRAP, "bootstrap", WS, "0", "0"]),
&none,
)
.await?;
if code != 0 {
return Err(Fault::Other(format!(
"{BOOTSTRAP_FAILED}: {}",
String::from_utf8_lossy(&err).trim()
)));
}
// The admitted workload owns a newly created workspace: no previous
// lease's manifest or build state is available, even with --region.
let plan = sync::plan(&self.tree, None);
let bytes = plan.upload_bytes(&self.tree);
let mut up = 0u64;
let filled = match store {
Some(st) if !plan.upload.is_empty() => {
let mut known = store::Known::load(
sync::git_path(&self.o.root, "sylphx/cas-known"),
&st.instance_key(),
);
let f = store::fill(
g,
st,
&mut known,
store::Local {
root: &self.o.root,
tree: &self.tree,
},
&plan,
WS,
USER,
)
.await?;
known.save(&self.tree);
up += f.sent;
if let Some(w) = &f.warning {
if !self.cache_warned.swap(true, Ordering::Relaxed) {
eprintln!(
"sylphx: warning: build store: {w}; sending those files directly"
);
}
}
f.paths
}
_ => Default::default(),
};
let direct: Vec<String> = plan
.upload
.iter()
.filter(|p| !filled.contains(*p))
.cloned()
.collect();
let tars = sync::tarballs(&self.o.root, &direct, &self.tree).map_err(Fault::Other)?;
let lists = sync::lists(&self.tree, &plan, &filled);
let s = format!("{WS}/.sylphx");
for (i, t) in tars.iter().enumerate() {
g.upload(&format!("{s}/in-{i:04}.tar.gz"), USER, t).await?;
up += t.len() as u64;
}
for (name, body) in [
("drop", lists.drop.as_bytes()),
("add", lists.add.as_bytes()),
("remove", &lists.remove[..]),
] {
if !body.is_empty() {
g.upload(&format!("{s}/{name}"), USER, body).await?;
up += body.len() as u64;
}
}
self.add(|st| {
st.bytes_up += up;
st.from_store = filled.len();
});
let full = match (self.o.fresh, plan.full) {
(true, _) => "2",
(false, true) => "1",
(false, false) => "0",
};
let (code, _, err) = g
.run(
USER,
&argv(&[
"/bin/sh",
"-c",
sync::APPLY,
"apply",
WS,
full,
if filled.is_empty() { "0" } else { "1" },
]),
&none,
)
.await?;
if code != 0 {
return Err(Fault::Other(format!(
"{APPLY_FAILED}: {}",
String::from_utf8_lossy(&err).trim()
)));
}
let secs = t.elapsed().as_secs_f64();
self.out.progress(&format!(
"synced {} files, {}, in {secs:.1} s{}{}",
plan.upload.len(),
human(up),
if filled.is_empty() {
String::new()
} else {
format!("; {} from the build store", filled.len())
},
if plan.remove.is_empty() {
String::new()
} else {
format!("; {} removed", plan.remove.len())
}
));
self.out.event(json!({
"type": "sync",
"full": plan.full,
"files": plan.upload.len(),
"removed": plan.remove.len(),
"bytes": bytes,
"bytes_up": up,
"from_store": filled.len(),
"duration_ms": t.elapsed().as_millis() as u64,
"workspace": "cold",
"volume": false,
}));
Ok(())
}
/// Puts the git dependencies the workspace lacks into its mirrors
/// ([`git_deps`]) and answers the command's environment for them. A
/// machine without git gets none and one warning: the build then fails
/// where it would have.
async fn send_git_deps(&self, g: &Guest) -> Result<BTreeMap<String, String>, Fault> {
if self.git_deps.is_empty() {
return Ok(BTreeMap::new());
}
let none = BTreeMap::new();
let names: Vec<String> = self.git_deps.iter().map(|(d, _)| d.name()).collect();
let mut args = argv(&["/bin/sh", "-c", git_deps::HAVE, "git-deps-have", WS]);
args.extend(names.iter().cloned());
let (code, out, err) = g.run(USER, &args, &none).await?;
if code == 3 {
eprintln!("sylphx: warning: the build machine has no git; git dependencies not sent");
return Ok(BTreeMap::new());
}
if code != 0 {
return Err(Fault::Other(format!(
"preparing the git dependencies failed: {}",
String::from_utf8_lossy(&err).trim()
)));
}
let missing: Vec<String> = String::from_utf8_lossy(&out)
.lines()
.filter(|l| names.iter().any(|n| n == l))
.map(str::to_string)
.collect();
let mut up = 0u64;
for (d, tgz) in &self.git_deps {
let n = d.name();
if !missing.contains(&n) {
continue;
}
let bytes = std::fs::read(tgz)
.map_err(|e| Fault::Other(format!("reading {}: {e}", tgz.display())))?;
g.upload(&format!("{WS}/{}/in/{n}.tgz", git_deps::DIR), USER, &bytes)
.await?;
up += bytes.len() as u64;
}
if !missing.is_empty() {
let mut args = argv(&["/bin/sh", "-c", git_deps::APPLY, "git-deps-apply", WS]);
args.extend(missing.iter().cloned());
let (code, _, err) = g.run(USER, &args, &none).await?;
if code != 0 {
return Err(Fault::Other(format!(
"adding the git dependencies failed: {}",
String::from_utf8_lossy(&err).trim()
)));
}
}
self.add(|s| s.bytes_up += up);
self.out.event(json!({
"type": "git_deps",
"deps": names.len(),
"sent": missing.len(),
"bytes_up": up,
}));
let deps: Vec<git_deps::Dep> = self.git_deps.iter().map(|(d, _)| d.clone()).collect();
Ok(git_deps::env(&deps, WS))
}
async fn send_source_history(&self, g: &Guest) -> Result<(), Fault> {
let Some((reference, bundle)) = &self.source_history else {
return Ok(());
};
g.upload(&format!("{WS}/.sylphx/source.bundle"), USER, bundle)
.await?;
let args = argv(&[
"/bin/sh",
"-c",
history::APPLY,
"source-history",
WS,
reference,
]);
let (code, _, err) = g.run(USER, &args, &BTreeMap::new()).await?;
if code != 0 {
return Err(Fault::Other(format!(
"installing source history: {}",
String::from_utf8_lossy(&err).trim()
)));
}
self.add(|s| s.bytes_up += bundle.len() as u64);
self.out.event(
json!({"type": "source_history", "reference": reference, "bytes_up": bundle.len()}),
);
Ok(())
}
/// Runs the command, streaming its output; `None` when `--timeout` hit.
/// A broken stream is reattached to the running command; the error says
/// whether the command had started when the run lost it.
async fn exec(
&self,
g: &mut Guest,
minted: &mut Instant,
lease: &str,
cache: BTreeMap<String, String>,
git_env: BTreeMap<String, String>,
) -> Result<Option<i32>, ExecFault> {
let mut args = vec![
"/bin/sh".to_string(),
"-c".into(),
RUN.into(),
"sylphx-run".into(),
WS.into(),
self.rel().to_string(),
"1".into(),
];
args.extend(self.o.command.iter().cloned());
// The command's own `--env` wins over the cache's and ours.
let mut env = cache;
if let Some(img) = &self.o.image {
// An image build gets the cache's mirror addresses, never its
// write tokens: no step of an image build needs sccache or Turbo.
env.retain(|k, _| image::forwarded(k));
env.entry("SYLPHX_OCI_MIRROR".into())
.or_insert_with(image::default_oci_mirror);
env.extend(img.env());
}
if let Some(sha) = sync::head(&self.o.root) {
env.insert("SYLPHX_BUILD_GIT_HEAD".into(), sha);
}
env.extend(git_env);
env.extend(self.o.env.clone());
if self.o.source_history.is_some() {
env.insert("SYLPHX_BUILD_SOURCE_HISTORY".into(), "1".into());
} else {
env.remove("SYLPHX_BUILD_SOURCE_HISTORY");
}
let mut p = g
.start(USER, &args, "", &env)
.await
.map_err(ExecFault::before)?;
let deadline = tokio::time::Instant::now() + self.o.timeout;
let mut tick = tokio::time::interval(KEEPALIVE);
tick.tick().await;
let mut reattached = false;
let result = loop {
tokio::select! {
ev = p.next() => match ev {
Ok(Some(Event::Start(pid))) => {
let g2 = g.clone();
self.held.with(|h| h.proc = Some((g2, pid)));
}
Ok(Some(Event::Stdout(b))) => self.out.stdout(&b),
Ok(Some(Event::Stderr(b))) => self.out.stderr(&b),
Ok(Some(Event::End(c))) => break Some(c),
Ok(Some(Event::Error(m))) => {
let f = Fault::Other(format!("the command could not run: {m}"));
return Err(ExecFault { started: p.pid.is_some(), fault: f });
}
Ok(None) | Err(Fault::Stream(_)) => {
// A proxy on the way restarted: follow the same
// process again rather than running it twice.
let broke = Instant::now();
if let Ok(t) = self.token(lease).await {
g.set_token(&t);
*minted = Instant::now();
}
// `--timeout` still holds while the stream is down.
let reattached_by = tokio::time::timeout_at(
deadline,
g.reattach_within(&mut p, REATTACH_BUDGET),
)
.await;
let Ok(reattach) = reattached_by else {
if let Some(pid) = p.pid {
let _ = g.signal(pid, "SIGKILL").await;
}
break None;
};
match reattach {
Ok(Reattach::Attached) => {
let gap = broke.elapsed().as_millis() as u64;
self.out.event(json!({"type": "reattached", "gap_ms": gap}));
if !reattached {
reattached = true;
self.out.progress(&format!(
"warning: the output stream broke; reattached to the running command after {:.1} s (output written meanwhile is not shown)",
gap as f64 / 1000.0
));
}
}
Ok(Reattach::NotRunning) => {
let f = Fault::Stream("the output stream broke and the command is no longer on the machine".into());
return Err(ExecFault { started: p.pid.is_some(), fault: f });
}
Err(f) => return Err(ExecFault { started: p.pid.is_some(), fault: f }),
}
}
Err(f) => return Err(ExecFault { started: p.pid.is_some(), fault: f }),
},
_ = tick.tick() => {
if minted.elapsed() >= TOKEN_RENEW {
if let Ok(t) = self.token(lease).await {
g.set_token(&t);
*minted = Instant::now();
}
}
let _ = g.touch(WS).await;
}
_ = tokio::time::sleep_until(deadline) => {
let pid = self.held.with(|h| h.proc.as_ref().map(|(_, p)| *p));
if let Some(pid) = pid {
let _ = g.signal(pid, "SIGKILL").await;
}
break None;
}
}
};
self.add(|s| s.bytes_down += p.bytes);
self.held.with(|h| h.proc = None);
Ok(result)
}
/// `build image --registry-auth`: puts the credentials on the machine as a
/// private file (0600 in a 0700 directory), and the file's values among
/// the secrets scrubbed from any text that leaves this process.
async fn send_auth(&self, g: &Guest) -> Result<(), Fault> {
let Some(file) = self.o.image.as_ref().and_then(|i| i.registry_auth.as_ref()) else {
return Ok(());
};
let bytes = std::fs::read(file).map_err(|e| {
Fault::Other(format!("reading --registry-auth {}: {e}", file.display()))
})?;
self.held.with(|h| {
h.secrets
.extend(image::auth_secrets(&String::from_utf8_lossy(&bytes)))
});
let none = BTreeMap::new();
let dir = image::private_dir();
let prep = r#"set -eu
umask 077
rm -rf "$1"
mkdir -m 700 "$1"
"#;
let (code, _, _) = g
.run(
USER,
&argv(&["/bin/sh", "-c", prep, "private", &dir]),
&none,
)
.await?;
if code != 0 {
return Err(Fault::Other(
"preparing the private directory failed".into(),
));
}
let path = image::auth_path();
g.upload(&path, USER, &bytes).await?;
let (code, _, _) = g.run(USER, &argv(&["chmod", "600", &path]), &none).await?;
if code != 0 {
return Err(Fault::Other(
"securing the registry credentials failed".into(),
));
}
Ok(())
}
/// Removes what [`Self::send_auth`] put on the machine; best effort (the
/// machine may be gone, and the next image build replaces the directory).
async fn drop_auth(&self, g: &Guest) {
if self
.o
.image
.as_ref()
.is_some_and(|i| i.registry_auth.is_some())
{
let _ = g
.run(
USER,
&argv(&["rm", "-rf", &image::private_dir()]),
&BTreeMap::new(),
)
.await;
}
}
/// `build image`: copies the script's metadata (and the `--oci-out`
/// archive) back and prints the digest, frontend and process sandbox.
async fn collect_image(&self, g: &Guest, lease: &str) -> Result<(), Outcome> {
let fail = |what: &str, f: Fault| Outcome::platform(format!("{what}: {}", f.text()), true);
let Some(bytes) = g
.download(&image::metadata_path(), USER)
.await
.map_err(|f| fail("copying the image metadata back", f))?
else {
return Err(Outcome::platform(
"the build ended without writing its metadata",
false,
));
};
let meta: Value = serde_json::from_slice(&bytes).map_err(|e| {
Outcome::platform(format!("the image metadata is not JSON: {e}"), false)
})?;
self.add(|s| s.bytes_down += bytes.len() as u64);
let dir = PathBuf::from(".sylphx/out").join(lease.rsplit('/').next().unwrap_or("run"));
let local = dir.join("image.json");
std::fs::create_dir_all(&dir)
.and_then(|_| std::fs::write(&local, &bytes))
.map_err(|e| Outcome::platform(format!("{}: {e}", local.display()), false))?;
if let Some(dest) = self.o.image.as_ref().and_then(|i| i.oci_out.as_ref()) {
let tar = g
.download(&image::oci_path(), USER)
.await
.map_err(|f| fail("copying the image archive back", f))?
.ok_or_else(|| Outcome::platform("the image archive was not written", false))?;
if let Some(parent) = dest.parent().filter(|p| !p.as_os_str().is_empty()) {
let _ = std::fs::create_dir_all(parent);
}
std::fs::write(dest, &tar)
.map_err(|e| Outcome::platform(format!("{}: {e}", dest.display()), false))?;
self.add(|s| s.bytes_down += tar.len() as u64);
}
for line in image::summary(&meta) {
if self.o.json {
continue;
}
println!("{line}");
}
self.add(|s| s.image = Some(meta));
Ok(())
}
/// Copies the `--artifact` matches back, checking each digest.
async fn collect(&self, g: &Guest, lease: &str) -> Result<(), String> {
let rel = if self.o.rel.is_empty() {
"."
} else {
&self.o.rel
};
let mut args = argv(&[
"/bin/bash",
"-O",
"globstar",
"-O",
"nullglob",
"-c",
ARTIFACTS,
"artifacts",
WS,
rel,
]);
args.extend(self.o.artifacts.iter().cloned());
let (_, listing, _) = g
.run(USER, &args, &BTreeMap::new())
.await
.map_err(|f| f.text().to_string())?;
let dir = self.o.out.clone().unwrap_or_else(|| {
PathBuf::from(".sylphx/out").join(lease.rsplit('/').next().unwrap_or("run"))
});
let mut n = 0usize;
for line in String::from_utf8_lossy(&listing).lines() {
let Some((digest, path)) = parse_sum(line) else {
continue;
};
let Some(local) = artifact_path(&dir, path) else {
continue;
};
let remote = format!("{WS}/tree/{rel}/{path}");
let bytes = g
.download(&remote, USER)
.await
.map_err(|f| f.text().to_string())?
.ok_or_else(|| format!("{path} disappeared before it was copied"))?;
if sync::sha256_hex(&bytes) != digest {
return Err(format!("{path} changed while it was copied"));
}
if let Some(parent) = local.parent() {
std::fs::create_dir_all(parent)
.map_err(|e| format!("{}: {e}", parent.display()))?;
}
std::fs::write(&local, &bytes).map_err(|e| format!("{}: {e}", local.display()))?;
self.add(|s| {
s.bytes_down += bytes.len() as u64;
s.artifacts += 1;
});
self.out.event(json!({
"type": "artifact",
"path": local.to_string_lossy(),
"digest": format!("sha256:{digest}"),
"bytes": bytes.len(),
}));
n += 1;
}
if n == 0 {
self.out.progress("no file matched --artifact");
}
Ok(())
}
}
/// The locked git dependencies a command run in `rel` can use, each packed
/// on this machine ([`git_deps::pack`]); the first that cannot be fetched is
/// the error.
fn prepare_git_deps(
root: &Path,
tree: &sync::Tree,
rel: &str,
) -> Result<Vec<(git_deps::Dep, PathBuf)>, String> {
let deps =
git_deps::locked(root, tree, rel).map_err(|e| format!("reading the lock files: {e}"))?;
if deps.is_empty() {
return Ok(Vec::new());
}
let cache = sync::git_path(root, "sylphx/git-deps")
.unwrap_or_else(|| std::env::temp_dir().join("sylphx-git-deps"));
std::fs::create_dir_all(&cache).map_err(|e| format!("preparing {}: {e}", cache.display()))?;
deps.into_iter()
.map(|d| git_deps::pack(&cache, &d).map(|p| (d, p)))
.collect()
}
/// How long preparing the machine (the toolchain install) may take.
const PROVISION_TIMEOUT: Duration = Duration::from_secs(15 * 60);
/// A refused API call as a platform failure; retryable as the API says, and
/// always when the API could not be reached.
fn api(what: &str, e: &sylphx::Error) -> Outcome {
let retryable = match e {
sylphx::Error::Api { retryable, .. } => *retryable,
sylphx::Error::Transport(_) => true,
_ => false,
};
Outcome::platform(format!("{what}: {}", why(e)), retryable)
}
/// An outcome's text with the cache token removed.
fn scrub(outcome: Outcome, secrets: &[String]) -> Outcome {
match outcome {
Outcome::Platform { reason, retryable } => Outcome::Platform {
reason: build_cache::scrub(&reason, secrets),
retryable,
},
Outcome::Lost(reason) => Outcome::Lost(build_cache::scrub(&reason, secrets)),
other => other,
}
}
/// A failure text that says a disk was full: ENOSPC from a guest tool, or
/// envd's 507 for an upload it could not finish ("not enough disk space
/// available", "not enough inodes available").
fn out_of_space(text: &str) -> bool {
text.contains("No space left on device")
|| text.contains("ENOSPC")
|| text.contains("507 Insufficient Storage")
|| text.contains("not enough disk space")
|| text.contains("not enough inodes")
}
/// How [`Run::sync`] names a failed bootstrap or apply step.
const BOOTSTRAP_FAILED: &str = "preparing the workspace failed";
const APPLY_FAILED: &str = "applying the sync failed";
/// A sync failure the run treats as a full workspace: one that says so, or
/// any failure of the bootstrap or apply step. Whatever stops these steps (a
/// truncated upload, a tool that died without saying ENOSPC) can leave the
/// tree and its manifest half-written, so the one retry starts from an
/// emptied workspace (a cold sync) rather than a second try on the same
/// state. A second failure is 125, retryable, as any loss before the start.
fn workspace_full(text: &str) -> bool {
out_of_space(text) || text.starts_with(BOOTSTRAP_FAILED) || text.starts_with(APPLY_FAILED)
}
fn argv(a: &[&str]) -> Vec<String> {
a.iter().map(|s| s.to_string()).collect()
}
/// One `sha256sum` line: the digest and the path. Escaped lines (a path with
/// a backslash or newline) are skipped.
fn parse_sum(line: &str) -> Option<(&str, &str)> {
if line.starts_with('\\') {
return None;
}
let (digest, rest) = line.split_once(" ")?;
(digest.len() == 64 && digest.bytes().all(|b| b.is_ascii_hexdigit())).then_some((digest, rest))
}
/// Where an artifact lands: its remote path under `dir`, with `.` and `..`
/// dropped so nothing lands outside `dir`.
fn artifact_path(dir: &Path, remote: &str) -> Option<PathBuf> {
let mut p = dir.to_path_buf();
let mut any = false;
for c in Path::new(remote).components() {
if let Component::Normal(s) = c {
p.push(s);
any = true;
}
}
any.then_some(p)
}
#[cfg(test)]
mod tests {
use super::*;
use sylphx_build_lease::BUILD_PACKAGES;
#[test]
fn a_failed_bootstrap_retries_before_the_command_starts() {
assert!(workspace_full(
"preparing the workspace failed: gzip: stdin: unexpected end of file"
));
assert!(workspace_full(
"preparing the workspace failed: gzip: No space left on device"
));
assert!(workspace_full("preparing the workspace failed: "));
assert!(!workspace_full("guest answered 503"));
let mut retries = Retries::default();
let failed = || Attempt::Lost("the workspace ran out of space".into());
assert!(matches!(retries.next(failed()), Step::Again { .. }));
match retries.next(failed()) {
Step::Done(outcome) => assert_eq!(outcome.code(), 125),
Step::Again { .. } => panic!("a bootstrap failure must not retry indefinitely"),
}
}
fn parse(args: &[&str]) -> Result<Opts, String> {
let mut root = Command::new("sylphx").arg(
Arg::new("output")
.long("output")
.short('o')
.global(true)
.default_value("table"),
);
root = root.subcommand(Command::new("build").subcommand(command()));
let m = root.try_get_matches_from(args).map_err(|e| e.to_string())?;
let (_, b) = m.subcommand().unwrap();
let (_, r) = b.subcommand().unwrap();
Opts::parse(r)
}
/// A fresh git work tree in a temporary directory, removed on drop, so a
/// test that parses a run over a tree never depends on the caller's
/// checkout (a copied source tree has no `.git`).
struct GitTree(std::path::PathBuf);
impl GitTree {
fn new() -> Self {
let p = std::env::temp_dir().join(format!(
"sylphx-build-tree-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&p).unwrap();
let ok = std::process::Command::new("git")
.arg("-C")
.arg(&p)
.args(["init", "-q"])
.status()
.unwrap()
.success();
assert!(ok, "git init in {}", p.display());
Self(p)
}
fn path(&self) -> &str {
self.0.to_str().unwrap()
}
}
impl Drop for GitTree {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
/// Run in a separate test process so Git tracing and CLI context are scoped
/// to this invocation, without changing other tests' environment or cwd.
#[tokio::test]
#[ignore = "subprocess entry point for the history ordering tests"]
async fn source_history_order_child() {
let root = std::env::var_os("SYLPHX_TEST_HISTORY_ROOT")
.expect("history ordering test must supply its scratch tree");
let root = PathBuf::from(root);
let base = std::env::var("SYLPHX_TEST_HISTORY_BASE").unwrap();
let mut opts = parse(&[
"sylphx",
"build",
"run",
root.to_str().unwrap(),
"--source-history",
&base,
"--",
"true",
])
.unwrap();
opts.dry_run = std::env::var_os("SYLPHX_TEST_HISTORY_DRY_RUN").is_some();
let client = Client::new(
sylphx::HttpTransport::builder()
.base_url(std::env::var("SYLPHX_TEST_HISTORY_API").unwrap())
.build()
.unwrap(),
);
let dry_run = opts.dry_run;
let result = run_opts(Ok(client), None, opts).await;
if dry_run {
assert!(result.is_ok());
} else {
assert!(matches!(result, Err(Failure::Exit(EXIT_PLATFORM))));
}
}
fn bundle_starts(trace: &Path) -> Vec<Value> {
std::fs::read_to_string(trace)
.unwrap_or_default()
.lines()
.filter_map(|line| serde_json::from_str::<Value>(line).ok())
.filter(|event| {
event["event"] == "start"
&& event["argv"].as_array().is_some_and(|args| {
args.windows(2)
.any(|pair| pair[0] == "bundle" && pair[1] == "create")
})
})
.collect()
}
/// The local HTTP fixture used by acquisition and source-history tests.
async fn read_http_request(socket: &mut tokio::net::TcpStream) -> (String, Value) {
use tokio::io::AsyncReadExt;
let mut request = Vec::new();
let mut chunk = [0; 8192];
loop {
let n = socket.read(&mut chunk).await.unwrap();
assert_ne!(n, 0);
request.extend_from_slice(&chunk[..n]);
if let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") {
let headers = String::from_utf8_lossy(&request[..end]);
let length = headers
.lines()
.find_map(|line| {
line.to_ascii_lowercase()
.strip_prefix("content-length:")
.map(|n| n.trim().parse::<usize>().unwrap())
})
.unwrap_or(0);
if request.len() >= end + 4 + length {
let line = headers.lines().next().unwrap().to_string();
let body = if length == 0 {
Value::Null
} else {
serde_json::from_slice(&request[end + 4..end + 4 + length]).unwrap()
};
return (line, body);
}
}
}
}
#[tokio::test]
async fn queued_run_and_image_leases_never_reserve_volumes_and_keep_region_binding() {
use tokio::io::AsyncWriteExt;
for image_build in [false, true] {
for region in [None, Some("hk")] {
let tree = GitTree::new();
let mut args = vec![
"sylphx",
"build",
if image_build { "image" } else { "run" },
tree.path(),
];
if let Some(region) = region {
args.extend(["--region", region]);
}
if !image_build {
args.extend(["--fresh", "--", "true"]);
}
let root = Command::new("sylphx")
.arg(
Arg::new("output")
.long("output")
.short('o')
.global(true)
.default_value("table"),
)
.subcommand(
Command::new("build")
.subcommand(command())
.subcommand(image_command()),
);
let matches = root.try_get_matches_from(args).unwrap();
let (_, build) = matches.subcommand().unwrap();
let (subcommand, matches) = build.subcommand().unwrap();
let opts = if subcommand == "image" {
Opts::parse_image(matches)
} else {
Opts::parse(matches)
}
.unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let accepted = region.unwrap_or("gra");
let server = tokio::spawn(async move {
// Only lease POST/GET is allowed. Any list/create/attach
// Volume request, including while queued, fails this fixture.
for attempt in 0..2 {
for create in [true, false] {
let (mut socket, _) = listener.accept().await.unwrap();
let (line, body) = read_http_request(&mut socket).await;
if create {
assert!(
line.starts_with("POST ") && line.contains("/leases "),
"{line}"
);
let wanted = if attempt == 0 {
region.unwrap_or("")
} else {
accepted
};
if wanted.is_empty() {
assert!(
body["spec"]
.get("region")
.is_none_or(|r| r.as_str() == Some("")),
"{body}"
);
} else {
assert_eq!(body["spec"]["region"], wanted, "{body}");
}
assert!(
body["spec"]
.get("volumes")
.is_none_or(|v| v.as_array().is_some_and(Vec::is_empty)),
"{body}"
);
assert_eq!(body["spec"]["shape"], "build-large");
} else {
assert!(
line.starts_with("GET ") && line.contains("/leases/l1 "),
"{line}"
);
}
let response = json!({
"name": "orgs/org_a/projects/prj_a/envs/env_a/leases/l1",
"spec": {"region": accepted},
"status": if create { json!({"state": "queued"}) } else { json!({"state": "ended", "endReason": "machine_lost"}) },
}).to_string();
socket.write_all(format!("HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response}", response.len()).as_bytes()).await.unwrap();
socket.shutdown().await.unwrap();
}
}
});
let client = Client::new(
sylphx::HttpTransport::builder()
.base_url(url)
.build()
.unwrap(),
);
let out = Out {
json: false,
quiet: true,
};
let held = Held::default();
let stats = Mutex::new(Stats::default());
let mut run = Run {
client: &client,
o: &opts,
out: &out,
held: &held,
stats: &stats,
parent: "orgs/org_a/projects/prj_a/envs/env_a".into(),
tree: BTreeMap::new(),
git_deps: Vec::new(),
source_history: None,
key: None,
cache_url: String::new(),
cache_warned: AtomicBool::new(false),
accepted_region: None,
};
for _ in 0..2 {
let acquired = tokio::time::timeout(Duration::from_secs(10), run.acquire())
.await
.unwrap();
assert!(matches!(acquired, Err(Attempt::Lost(_))));
assert_eq!(run.accepted_region.as_deref(), Some(accepted));
assert_eq!(run.region(), accepted);
}
server.await.unwrap();
}
}
}
async fn history_before_acquire(base: &str, dry_run: bool, expect_attempts: usize) {
use tokio::io::AsyncWriteExt;
let tree = GitTree::new();
let git = |args: &[&str]| {
let output = std::process::Command::new("git")
.current_dir(&tree.0)
.args([
"-c",
"user.name=test",
"-c",
"user.email=test@example.invalid",
"-c",
"commit.gpgsign=false",
])
.args(args)
.output()
.unwrap();
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
};
std::fs::write(tree.0.join("file"), "committed\n").unwrap();
git(&["add", "file"]);
git(&["commit", "-qm", "source"]);
git(&["update-ref", "refs/remotes/origin/main", "HEAD"]);
std::fs::write(tree.0.join("file"), "dirty\n").unwrap();
let trace = tree.0.join(".git/source-history-trace.jsonl");
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let starts_at_acquire = std::sync::Arc::new(parking_lot::Mutex::new(Vec::new()));
let seen = starts_at_acquire.clone();
let source = tree.0.clone();
let traced = trace.clone();
let server = tokio::spawn(async move {
loop {
let (mut socket, _) = listener.accept().await.unwrap();
let (line, _) = read_http_request(&mut socket).await;
let (status, body) = if line.starts_with("POST ") && line.contains("/leases ") {
let starts = bundle_starts(&traced);
let events = std::fs::read_to_string(&traced).unwrap_or_default();
let completed = starts
.iter()
.filter(|start| {
events.lines().any(|line| {
serde_json::from_str::<Value>(line).is_ok_and(|event| {
event["event"] == "exit"
&& event["sid"] == start["sid"]
&& event["code"] == 0
})
})
})
.count();
seen.lock().push(completed);
// Any attempted re-preparation after this acquisition must
// fail: only the already-built bundle can be reused.
let reference = source.join(".git/refs/remotes/origin/main");
if reference.exists() {
std::fs::remove_file(reference).unwrap();
}
(
200,
json!({"name": "orgs/org_a/projects/prj_a/envs/env_a/leases/l1",
"status": {"state": "ended", "endReason": "machine_lost"}}),
)
} else {
panic!("unexpected API request: {line}");
};
let body = body.to_string();
socket.write_all(format!(
"HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len(),
).as_bytes()).await.unwrap();
socket.shutdown().await.unwrap();
}
});
let mut child = tokio::process::Command::new(std::env::current_exe().unwrap());
child
.args([
"--exact",
"build_run::tests::source_history_order_child",
"--ignored",
"--nocapture",
])
.current_dir(&tree.0)
.env("SYLPHX_TEST_HISTORY_ROOT", &tree.0)
.env("SYLPHX_TEST_HISTORY_BASE", base)
.env("SYLPHX_TEST_HISTORY_API", url)
.env(
crate::context::ENV_VAR,
"orgs/org_a/projects/prj_a/envs/env_a",
)
.env("GIT_TRACE2_EVENT", &trace)
.env_remove("SYLPHX_TEST_HISTORY_DRY_RUN")
.kill_on_drop(true);
if dry_run {
child.env("SYLPHX_TEST_HISTORY_DRY_RUN", "1");
}
let result = tokio::time::timeout(Duration::from_secs(30), child.output()).await;
server.abort();
let output = result.unwrap().unwrap();
assert!(
output.status.success(),
"{}\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
if expect_attempts == 0 && !dry_run {
assert!(String::from_utf8_lossy(&output.stderr).contains("preparing source history"));
}
assert_eq!(*starts_at_acquire.lock(), vec![1; expect_attempts]);
assert_eq!(
bundle_starts(&trace).len(),
usize::from(expect_attempts > 0)
);
assert_eq!(
std::fs::read_to_string(tree.0.join("file")).unwrap(),
"dirty\n"
);
}
#[tokio::test]
async fn source_history_finishes_before_acquire_and_is_not_reprepared_on_retry() {
history_before_acquire("origin/main", false, 2).await;
}
#[tokio::test]
async fn source_history_failure_never_acquires_a_lease() {
history_before_acquire("origin/missing", false, 0).await;
}
#[tokio::test]
async fn dry_run_does_not_prepare_source_history_or_acquire() {
history_before_acquire("origin/missing", true, 0).await;
}
#[test]
fn source_history_is_rejected_for_no_tree_runs() {
assert!(parse(&[
"sylphx",
"build",
"run",
"--no-tree",
"--source-history",
"origin/main",
"--",
"true",
])
.is_err());
}
/// Runs [`RUN`] the way the guest does (`sh -c RUN … -- <command>`) in a
/// scratch workspace, with `rustup` and `curl` replaced by stubs: the stub
/// rustup answers `which` only under `ok_home`, the stub curl succeeds
/// when `mirror_up`. The workspace starts with a stale Cargo config from
/// an earlier run. Returns the command's stdout.
fn run_prelude(ok_home: &str, mirror: Option<&str>, mirror_up: bool, cmd: &str) -> String {
run_prelude_env(ok_home, mirror, mirror_up, cmd, &[]).0
}
/// [`run_prelude`] with extra environment, returning stdout and stderr.
/// `sccache` is a stub too, so no real server starts: `--start-server`
/// exits `$STUB_SCCACHE_START` (default 0) and says why on stderr.
fn run_prelude_env(
ok_home: &str,
mirror: Option<&str>,
mirror_up: bool,
cmd: &str,
env: &[(&str, &str)],
) -> (String, String) {
let base = std::env::temp_dir().join(format!(
"sylphx-build-run-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let (ws, bin) = (base.join("ws"), base.join("bin"));
std::fs::create_dir_all(ws.join("tree/crate")).unwrap();
std::fs::create_dir_all(ws.join(".sylphx")).unwrap();
// A replacement left by an earlier run on this workspace.
std::fs::create_dir_all(ws.join(".cargo")).unwrap();
std::fs::write(ws.join(".cargo/config.toml"), "stale\n").unwrap();
std::fs::create_dir_all(&bin).unwrap();
let ok_home = ok_home.replace("$W", ws.to_str().unwrap());
let stubs = [
(
"rustup",
format!(
"#!/bin/sh\n[ \"$1\" = which ] && [ \"$RUSTUP_HOME\" = \"{ok_home}\" ] && exit 0\n[ \"$1\" = which ] && exit 1\nexit 0\n"
),
),
(
"curl",
format!("#!/bin/sh\nexit {}\n", if mirror_up { 0 } else { 7 }),
),
(
"sccache",
"#!/bin/sh\nif [ \"$1\" = --start-server ]; then\n echo 'sccache: Starting the server...'\n [ \"${STUB_SCCACHE_START:-0}\" = 0 ] && exit 0\n printf '\\nsccache: error: Server startup failed: cache storage failed to read\\n\\nContext:\\n' >&2\n exit \"$STUB_SCCACHE_START\"\nfi\nexit 0\n"
.to_string(),
),
];
for (name, body) in stubs {
let f = bin.join(name);
std::fs::write(&f, body).unwrap();
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o755)).unwrap();
}
let mut c = std::process::Command::new("sh");
c.args(["-c", RUN, "sylphx-run"])
.arg(&ws)
.args(["crate", "1", "sh", "-c", cmd])
.env_clear()
.env("PATH", format!("{}:/usr/bin:/bin", bin.display()));
if let Some(m) = mirror {
c.env("SYLPHX_CRATES_MIRROR", m);
}
c.envs(env.iter().copied());
let out = c.output().unwrap();
let _ = std::fs::remove_dir_all(&base);
assert!(
out.status.success(),
"{}",
String::from_utf8_lossy(&out.stderr)
);
(
String::from_utf8(out.stdout).unwrap(),
String::from_utf8(out.stderr).unwrap(),
)
}
#[test]
fn incremental_compilation_does_not_use_sccache() {
let (out, err) = run_prelude_env(
"/opt/rustup",
None,
false,
"echo \"${CARGO_INCREMENTAL}:${RUSTC_WRAPPER:-}\"",
&[("CARGO_INCREMENTAL", "1"), ("STUB_SCCACHE_START", "2")],
);
assert_eq!(out, "1:\n");
assert!(!err.contains("sylphx: warning: the build cache"), "{err}");
}
#[test]
fn lease_owned_targets_use_sccache_without_incremental_by_default() {
let show = "echo \"${CARGO_INCREMENTAL}:${RUSTC_WRAPPER:-}\"";
let (out, _) = run_prelude_env("/opt/rustup", None, false, show, &[]);
assert_eq!(out, "0:\n");
let (out, _) = run_prelude_env(
"/opt/rustup",
None,
false,
show,
&[("SCCACHE_WEBDAV_ENDPOINT", "http://build-cache.test/sccache")],
);
assert_eq!(out, "0:sccache\n");
}
#[test]
fn a_build_cache_that_does_not_answer_never_fails_the_build() {
let wrapper = "echo \"[${RUSTC_WRAPPER:-}]\"";
let webdav = ("SCCACHE_WEBDAV_ENDPOINT", "http://build-cache.test/sccache");
// The cache answers: compiles go through sccache, nothing is said.
let (out, err) = run_prelude_env(
"/opt/rustup",
None,
false,
wrapper,
&[webdav, ("CARGO_INCREMENTAL", "0")],
);
assert_eq!(out, "[sccache]\n");
assert!(!err.contains("sylphx: warning: the build cache"), "{err}");
// sccache's server refuses to start (its backend is down): the run
// compiles without it and says so once, with sccache's reason.
let (out, err) = run_prelude_env(
"/opt/rustup",
None,
false,
wrapper,
&[
webdav,
("CARGO_INCREMENTAL", "0"),
("STUB_SCCACHE_START", "2"),
],
);
assert_eq!(out, "[]\n");
assert_eq!(
err.matches("sylphx: warning: the build cache did not answer, so this run compiles without it: sccache: error: Server startup failed: cache storage failed to read\n").count(),
1,
"{err}"
);
// A caller's own wrapper is left alone and no server is started.
let (out, _) = run_prelude_env(
"/opt/rustup",
None,
false,
wrapper,
&[
webdav,
("STUB_SCCACHE_START", "2"),
("RUSTC_WRAPPER", "mine"),
],
);
assert_eq!(out, "[mine]\n");
}
#[test]
fn the_template_toolchain_is_used_when_it_serves_the_tree() {
// The guest inherits no RUSTUP_HOME (envd passes only PATH, HOME,
// USER and LOGNAME): the template's home is still chosen when it
// holds the tree's toolchain.
assert_eq!(
run_prelude("/opt/rustup", None, false, "echo $RUSTUP_HOME"),
"/opt/rustup\n"
);
// It does not: the workspace's rustup home, installed there.
let home = run_prelude("$W/rustup", None, false, "echo $RUSTUP_HOME");
assert!(home.trim_end().ends_with("/ws/rustup"), "{home}");
}
#[test]
fn crates_come_through_the_mirror_only_while_it_answers() {
let m = "sparse+http://build-cache.env-1.svc.cluster.local/crates/index/";
let cfg = run_prelude("/opt/rustup", Some(m), true, "cat ../../.cargo/config.toml");
assert_eq!(
cfg,
format!(
"[source.crates-io]\nreplace-with = \"sylphx-mirror\"\n\n[source.sylphx-mirror]\nregistry = \"{m}\"\n"
)
);
// Unreachable or not offered: no replacement, Cargo goes direct.
for (mirror, up) in [(Some(m), false), (None, true)] {
let out = run_prelude(
"/opt/rustup",
mirror,
up,
"test -e ../../.cargo/config.toml && echo present || echo absent",
);
assert_eq!(out, "absent\n");
}
}
#[test]
fn lease_egress_is_the_cache_gateway_plus_allow_hosts_only() {
// No internet package host by default: the mirrors serve them.
assert_eq!(allowed_domains(&[]), ["build-cache.sylphx.net"]);
for h in [
"index.crates.io",
"static.crates.io",
"static.rust-lang.org",
"registry.npmjs.org",
"github.com",
"codeload.github.com",
"objects.githubusercontent.com",
] {
assert!(!allowed_domains(&[]).contains(&h.to_string()), "{h}");
}
// A host the caller names is the opt-in.
let d = allowed_domains(&["registry.npmjs.org".to_string()]);
assert_eq!(d, ["build-cache.sylphx.net", "registry.npmjs.org"]);
}
#[test]
fn an_image_build_leases_with_no_internet_package_host() {
let root = Command::new("sylphx")
.arg(
Arg::new("output")
.long("output")
.short('o')
.global(true)
.default_value("table"),
)
.subcommand(Command::new("build").subcommand(image_command()));
// A work tree of its own: the lease's copy has no `.git`.
let tree = std::env::temp_dir().join(format!("sylphx-image-egress-{}", std::process::id()));
std::fs::create_dir_all(&tree).unwrap();
assert!(std::process::Command::new("git")
.arg("-C")
.arg(&tree)
.args(["init", "-q"])
.status()
.unwrap()
.success());
let m = root
.try_get_matches_from([
"sylphx",
"build",
"image",
tree.to_str().unwrap(),
"--allow-host",
"Proxy.Example.com",
])
.unwrap();
let (_, b) = m.subcommand().unwrap();
let (_, r) = b.subcommand().unwrap();
let o = Opts::parse_image(r).unwrap();
let _ = std::fs::remove_dir_all(&tree);
assert_eq!(
lease_domains(&o),
["build-cache.sylphx.net", "proxy.example.com"]
);
assert!(o.command[5].ends_with("sylphx-image-build"));
// `build run` keeps its preset.
assert_eq!(allowed_domains(&[]).len(), BUILD_PACKAGES.len());
}
// The guest scripts run for real under /bin/sh in a scratch workspace;
// rustup is a fake placed where the script puts the workspace's cargo/bin.
struct Ws(PathBuf);
/// The toolchain the fake rustup installs.
const TC: &str = "1.99.0-x86_64-unknown-linux-gnu";
impl Ws {
fn new(tag: &str) -> Ws {
let d = std::env::temp_dir().join(format!(
"sylphx-build-run-{tag}-{}-{}",
std::process::id(),
sylphx_build_lease::guest::rand_u64()
));
std::fs::create_dir_all(d.join("tree")).unwrap();
std::fs::create_dir_all(d.join(".sylphx")).unwrap();
std::fs::create_dir_all(d.join("cargo/bin")).unwrap();
Ws(d)
}
fn file(&self, rel: &str, body: &str) {
let p = self.0.join(rel);
std::fs::create_dir_all(p.parent().unwrap()).unwrap();
std::fs::write(&p, body).unwrap();
}
/// A fake rustup over real toolchain directories: `which rustc`
/// answers `<home>/toolchains/<TC>/bin/rustc` when it is there, with
/// the template's home (`/opt/rustup`) kept in `image-rustup` of the
/// workspace; `toolchain install` writes the toolchain (whose rustc
/// runs only while its driver library is intact) unless `offline`
/// exists, and fails through the cache tier while `tier-down` exists.
/// Like rustup, it leaves a toolchain directory that is already there
/// alone. Each download adds a line to `fetches`.
fn rustup(&self, fail_install: bool) {
let w = self.0.display();
if fail_install {
self.file("offline", "");
}
let body = format!(
r#"#!/bin/sh
H=${{RUSTUP_HOME:-$HOME/.rustup}}
[ "$H" = /opt/rustup ] && H='{w}/image-rustup'
T="$H/toolchains/{TC}"
case "$1 $2" in
which*) [ -x "$T/bin/rustc" ] && echo "$T/bin/rustc" ;;
"toolchain install")
[ -e '{w}/offline' ] && {{ echo 'error: could not download channel-rust-1.99.0.toml.sha256 (Connection timed out)' >&2; exit 1; }}
[ -n "$RUSTUP_DIST_SERVER" ] && [ -e '{w}/tier-down' ] && {{ echo 'error: 502 from the tier' >&2; exit 1; }}
[ -d "$T" ] && exit 0
mkdir -p "$T/bin" "$T/lib/rustlib"
printf 'rustc\n' > "$T/lib/rustlib/components"
printf 'driver\n' > "$T/lib/librustc_driver.so"
printf '#!/bin/sh\n[ "$(cat "$(dirname "$0")/../lib/librustc_driver.so")" = driver ] || {{ echo "error while loading shared libraries: librustc_driver.so: cannot read file data" >&2; exit 127; }}\necho "rustc 1.99.0"\n' > "$T/bin/rustc"
printf '#!/bin/sh\necho "cargo 1.99.0"\n' > "$T/bin/cargo"
chmod +x "$T/bin/rustc" "$T/bin/cargo"
echo "${{RUSTUP_DIST_SERVER:-direct}}" >> '{w}/fetches' ;;
"toolchain uninstall") rm -rf "$H/toolchains/$3" ;;
default*) echo 'error: no network' >&2; exit 1 ;;
esac
"#
);
self.file("cargo/bin/rustup", &body);
use std::os::unix::fs::PermissionsExt;
let p = self.0.join("cargo/bin/rustup");
std::fs::set_permissions(&p, std::fs::Permissions::from_mode(0o755)).unwrap();
}
/// Bakes the toolchain into the template's home, as the image does.
fn bake_image(&self) {
// Through sh, not exec'd: a test thread forking while another
// just wrote its fake would make the exec fail with ETXTBSY.
let out = std::process::Command::new("/bin/sh")
.arg(self.0.join("cargo/bin/rustup"))
.args(["toolchain", "install"])
.env_clear()
.env("PATH", SYS_PATH)
.env("RUSTUP_HOME", "/opt/rustup")
.output()
.unwrap();
assert!(out.status.success());
let _ = std::fs::remove_file(self.0.join("fetches"));
}
/// The toolchain directory under a rustup home of the workspace.
fn tc(&self, home: &str) -> PathBuf {
self.0.join(home).join("toolchains").join(TC)
}
/// How many downloads the fake rustup made.
fn fetches(&self) -> usize {
std::fs::read_to_string(self.0.join("fetches"))
.map(|s| s.lines().count())
.unwrap_or(0)
}
fn sh(&self, script: &str, rel: &str, extra: &[&str]) -> (i32, String) {
self.sh_env(script, rel, extra, &[])
}
fn sh_env(
&self,
script: &str,
rel: &str,
extra: &[&str],
env: &[(&str, &str)],
) -> (i32, String) {
let out = std::process::Command::new("/bin/sh")
.arg("-c")
.arg(script)
.arg("sylphx-test")
.arg(&self.0)
.arg(rel)
.args(extra)
.env_clear()
.env("PATH", "/usr/bin:/bin")
.envs(env.iter().copied())
.output()
.unwrap();
(
out.status.code().unwrap_or(-1),
String::from_utf8_lossy(&out.stderr).into_owned(),
)
}
}
impl Ws {
/// Runs [`RUN`] in `tree/<rel>` with `PATH` and extra `env`;
/// answers the status, stdout and stderr.
fn run(
&self,
rel: &str,
path: &str,
env: &[(&str, &str)],
cmd: &str,
) -> (i32, String, String) {
let out = std::process::Command::new("/bin/sh")
.args(["-c", RUN, "sylphx-run"])
.arg(&self.0)
.args([rel, "0", "sh", "-c", cmd])
.env_clear()
.env("PATH", path)
.envs(env.iter().copied())
.output()
.unwrap();
(
out.status.code().unwrap_or(-1),
String::from_utf8_lossy(&out.stdout).into_owned(),
String::from_utf8_lossy(&out.stderr).into_owned(),
)
}
/// A `bin` directory holding only links to `tools` (and `sh`).
fn bin(&self, name: &str, tools: &[&str]) -> String {
let d = self.0.join(name);
std::fs::create_dir_all(&d).unwrap();
for t in tools.iter().chain(&["sh"]) {
let src = ["/usr/bin", "/bin"]
.iter()
.map(|b| Path::new(b).join(t))
.find(|p| p.exists())
.unwrap_or_else(|| panic!("{t} is not installed"));
let _ = std::os::unix::fs::symlink(src, d.join(t));
}
d.display().to_string()
}
}
const SYS_PATH: &str = "/usr/bin:/bin";
#[test]
fn the_tree_is_a_git_work_tree_naming_the_local_head() {
let w = Ws::new("git-tree");
w.file("tree/crate/a.txt", "a\n");
w.file("tree/b.txt", "b\n");
std::os::unix::fs::symlink("../target", w.0.join("tree/target")).unwrap();
let (code, out, err) = w.run(
"crate",
SYS_PATH,
&[("SYLPHX_BUILD_GIT_HEAD", "0123abc")],
"git rev-parse --is-inside-work-tree && git log -1 --format=%s && git ls-files --full-name :/",
);
assert_eq!(code, 0, "{err}");
assert_eq!(
out,
"true\nsylphx build run of 0123abc\nb.txt\ncrate/a.txt\n"
);
assert!(!err.contains("warning"), "{err}");
}
#[test]
fn source_history_run_keeps_the_imported_head_and_dirty_files() {
let w = Ws::new("source-history-run");
w.file("tree/file", "base\n");
let (code, head, err) = w.run(".", SYS_PATH, &[], "git rev-parse HEAD");
assert_eq!(code, 0, "{err}");
w.file("tree/file", "dirty\n");
let (code, out, err) = w.run(
".",
SYS_PATH,
&[("SYLPHX_BUILD_SOURCE_HISTORY", "1")],
"git rev-parse HEAD && git diff --name-only && cat file",
);
assert_eq!(code, 0, "{err}");
assert_eq!(out, format!("{head}file\ndirty\n"));
}
#[test]
fn a_warm_workspace_keeps_one_commit_of_the_current_tree() {
let w = Ws::new("git-warm");
w.file("tree/keep.txt", "keep\n");
w.file("tree/edit.txt", "v1\n");
let (code, _, err) = w.run(".", SYS_PATH, &[], "true");
assert_eq!(code, 0, "{err}");
w.file("tree/edit.txt", "v2\n");
let (code, out, err) = w.run(
".",
SYS_PATH,
&[],
"git rev-list --count --all && git log -1 --format=%s && git status --porcelain && git show HEAD:edit.txt && git ls-files && test -z \"$(git reflog)\" && echo no-reflog",
);
assert_eq!(code, 0, "{err}");
// One commit however many runs and no reflog keeping the replaced ones:
// the warm repository does not grow per run.
assert_eq!(
out,
"1\nsylphx build run of an unknown commit\nv2\nedit.txt\nkeep.txt\nno-reflog\n"
);
assert!(!err.contains("warning"), "{err}");
}
#[test]
fn a_failed_git_step_warns_and_the_command_keeps_its_exit_code() {
let w = Ws::new("git-fail");
w.file("tree/a.txt", "a\n");
w.file(
"fake/git",
"#!/bin/sh\necho 'fatal: disk full' >&2\nexit 128\n",
);
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(w.0.join("fake/git"), std::fs::Permissions::from_mode(0o755))
.unwrap();
let path = format!("{}:{SYS_PATH}", w.0.join("fake").display());
let (code, out, err) = w.run(".", &path, &[], "echo ran; exit 7");
assert_eq!((code, out.as_str()), (7, "ran\n"), "{err}");
assert_eq!(err.matches("sylphx: warning:").count(), 1, "{err}");
assert!(err.contains("fatal: disk full"), "{err}");
}
#[test]
fn a_machine_without_git_warns_and_runs_the_command() {
let w = Ws::new("git-none");
w.file("tree/a.txt", "a\n");
let path = w.bin("nogit", &["mkdir", "rm", "tail"]);
let (code, out, err) = w.run(".", &path, &[], "echo ran");
assert_eq!((code, out.as_str()), (0, "ran\n"), "{err}");
assert_eq!(err.matches("sylphx: warning:").count(), 1, "{err}");
assert!(err.contains("has no git"), "{err}");
assert!(!w.0.join("tree/.git").exists());
}
fn bootstrap(w: &Ws, path: &str, args: &[&str]) -> (i32, String, String) {
let out = std::process::Command::new("/bin/sh")
.args(["-c", BOOTSTRAP, "bootstrap"])
.arg(w.0.join("ws"))
.args(args)
.env_clear()
.env("PATH", path)
.output()
.unwrap();
(
out.status.code().unwrap_or(-1),
String::from_utf8_lossy(&out.stdout).into_owned(),
String::from_utf8_lossy(&out.stderr).into_owned(),
)
}
#[test]
fn the_bootstrap_floor_matches_the_named_limits() {
use sylphx_build_lease::scripts::{WORKSPACE_MIN_FREE_GIB, WORKSPACE_MIN_FREE_PERCENT};
assert!(BOOTSTRAP.contains(&format!(
"MIN_GIB={WORKSPACE_MIN_FREE_GIB} MIN_PCT={WORKSPACE_MIN_FREE_PERCENT}"
)));
assert!(out_of_space("gzip: stdout: No space left on device"));
assert!(!out_of_space("guest answered 503"));
}
/// Any failed apply and envd's 507 for a cut-short upload retry once on
/// a fresh lease-owned workspace; other faults keep their own verdict.
#[test]
fn a_failed_apply_or_a_cut_short_upload_counts_as_a_full_workspace() {
assert!(workspace_full(
"applying the sync failed: gzip: stdin: unexpected end of file"
));
assert!(workspace_full(&format!("{APPLY_FAILED}: tar: short read")));
let envd = r#"guest answered 507 Insufficient Storage: {"code":507,"message":"not enough disk space available"}"#;
assert!(out_of_space(envd) && workspace_full(envd));
assert!(out_of_space("not enough inodes available: no space"));
assert!(!workspace_full("guest answered 503 Service Unavailable"));
assert!(!workspace_full("preparing the git dependencies failed: x"));
}
/// A gzip tarball of one file, `name`, as the client sends it, with
/// incompressible content so that a cut-off copy ends mid-stream.
fn tarball(w: &Ws, name: &str, bytes: usize) -> Vec<u8> {
let mut x: u64 = 0x9e37_79b9_7f4a_7c15;
let body: String = (0..bytes)
.map(|_| {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
char::from(b'!' + (x % 90) as u8)
})
.collect();
w.file(&format!("src/{name}"), &body);
let out = std::process::Command::new("tar")
.args(["-czf", "-", "-C"])
.arg(w.0.join("src"))
.arg(name)
.output()
.unwrap();
assert!(out.status.success());
out.stdout
}
fn apply(w: &Ws, full: &str) -> (bool, String) {
let out = std::process::Command::new("/bin/sh")
.args(["-c", sync::APPLY, "apply"])
.arg(w.0.join("ws"))
.args([full, "0"])
.env_clear()
.env("PATH", SYS_PATH)
.output()
.unwrap();
(
out.status.success(),
String::from_utf8_lossy(&out.stderr).into_owned(),
)
}
/// 2026-10-06 21:54Z: two runs on a full workspace had envd answer 507
/// partway through an upload, which leaves the part it wrote; at 21:59Z
/// the next run pruned `target/`, sent its own tarball, and its apply
/// read the leftover too: `gzip: stdin: unexpected end of file`. The
/// bootstrap now clears every leftover before the sync, so the same
/// state applies cleanly; a leftover `remove` list would otherwise have
/// deleted a tree file without failing.
#[test]
fn a_truncated_upload_left_by_an_earlier_run_never_reaches_the_next_apply() {
let w = Ws::new("ws-leftover");
let fresh = tarball(&w, "new.rs", 64);
let cut = tarball(&w, "old.rs", 256 << 10);
let leftover = |w: &Ws| {
w.file("ws/tree/keep.rs", "fn keep() {}\n");
let s = w.0.join("ws/.sylphx");
std::fs::create_dir_all(&s).unwrap();
// keep.rs is input, not a file generated by an earlier build.
std::fs::write(
s.join("manifest"),
format!("{}\n{} 100644 keep.rs\n", sync::HEADER, "0".repeat(64)),
)
.unwrap();
std::fs::write(s.join("in-0003.tar.gz"), &cut[..cut.len() / 2]).unwrap();
std::fs::write(s.join("remove"), b"keep.rs\0").unwrap();
};
let send = |w: &Ws| {
std::fs::write(w.0.join("ws/.sylphx/in-0000.tar.gz"), &fresh).unwrap();
};
// The failure as it was: the apply meets the cut-off tarball.
leftover(&w);
send(&w);
let (ok, err) = apply(&w, "0");
assert!(!ok);
assert!(err.contains("unexpected end of file"), "{err}");
assert!(workspace_full(&format!("{APPLY_FAILED}: {}", err.trim())));
// The same leftovers with the bootstrap first: a clean apply.
let w = Ws::new("ws-leftover-fixed");
leftover(&w);
let (code, _, err) = bootstrap(&w, SYS_PATH, &["0", "0"]);
assert_eq!(code, 0, "{err}");
send(&w);
let (ok, err) = apply(&w, "0");
assert!(ok, "{err}");
assert!(w.0.join("ws/tree/new.rs").exists());
assert!(w.0.join("ws/tree/keep.rs").exists(), "no stale remove list");
assert!(!w.0.join("ws/tree/old.rs").exists());
}
/// Every file a sync stages before its apply is removed by the
/// bootstrap; the manifest, the tree and the caches stay.
#[test]
fn the_bootstrap_clears_every_sync_leftover_and_keeps_the_workspace() {
use sylphx_build_lease::scripts::SYNC_LEFTOVERS;
let w = Ws::new("ws-clear");
for l in SYNC_LEFTOVERS {
let p = l.replace('*', "0007");
if p.ends_with("fill") || p.ends_with("/in") {
w.file(&format!("ws/{p}/x"), "x");
} else {
w.file(&format!("ws/{p}"), "x");
}
}
w.file("ws/.sylphx/manifest", "a.rs\n");
w.file("ws/tree/a.rs", "fn a() {}\n");
w.file("ws/target/debug/a", "x");
w.file("ws/git-deps/0123456789abcdef.git/HEAD", "x");
let (code, _, err) = bootstrap(&w, SYS_PATH, &["0", "0"]);
assert_eq!(code, 0, "{err}");
for l in SYNC_LEFTOVERS {
let p = l.replace('*', "0007");
assert!(!w.0.join("ws").join(&p).exists(), "{p} is left");
}
for keep in [
".sylphx/manifest",
".sylphx/manifest.gz",
"tree/a.rs",
"target/debug/a",
"git-deps/0123456789abcdef.git/HEAD",
] {
assert!(w.0.join("ws").join(keep).exists(), "{keep} is gone");
}
}
#[test]
fn cargo_uses_the_persistent_target_unless_the_caller_says_otherwise() {
let w = Ws::new("cargo-env");
w.file("tree/a.txt", "a\n");
let show =
r#"echo "inc=${CARGO_INCREMENTAL-unset} debug=${CARGO_PROFILE_DEV_DEBUG-unset}""#;
let (code, out, err) = w.run(".", SYS_PATH, &[], show);
assert_eq!((code, out.as_str()), (0, "inc=1 debug=0\n"), "{err}");
let (code, out, err) = w.run(
".",
SYS_PATH,
&[("CARGO_INCREMENTAL", "1"), ("CARGO_PROFILE_DEV_DEBUG", "")],
show,
);
assert_eq!((code, out.as_str()), (0, "inc=1 debug=\n"), "{err}");
}
impl Drop for Ws {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
#[test]
fn a_failed_toolchain_install_stops_a_rust_tree_before_its_command() {
let w = Ws::new("tc-fail");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(true);
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 4, "{err}");
assert!(err.contains("could not be installed"), "{err}");
assert!(
err.contains("Connection timed out"),
"the log's tail is shown: {err}"
);
// Whatever the status, a failed preparation is a platform failure.
let o = Outcome::platform(provision_failure(code, err.as_bytes()), true);
assert_eq!(o.code(), 125);
assert!(
matches!(&o, Outcome::Platform { reason, .. } if reason.contains("Connection timed out"))
);
}
#[test]
fn a_toolchain_file_above_the_run_directory_counts_as_a_rust_tree() {
let w = Ws::new("tc-sub");
w.file(
"tree/rust-toolchain.toml",
"[toolchain]\nchannel = \"1.99.0\"\n",
);
w.file("tree/sub/dir/x.txt", "x");
w.rustup(true);
assert_eq!(w.sh(PROVISION, "sub/dir", &[]).0, 4);
}
#[test]
fn a_failed_install_does_not_stop_a_tree_without_rust() {
let w = Ws::new("tc-none");
w.file("tree/package.json", "{}");
w.rustup(true);
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
}
#[test]
fn a_successful_install_provisions() {
let w = Ws::new("tc-ok");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 0, "{err}");
assert!(w.tc("rustup").join(".sylphx-sha256").exists());
// Already installed and intact: nothing to do.
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
assert_eq!(w.fetches(), 1);
}
#[test]
fn the_template_toolchain_serves_a_tree_it_holds_with_no_workspace_copy() {
let w = Ws::new("tc-image");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
w.bake_image();
// A lease's guest inherits no RUSTUP_HOME: the template's is chosen,
// nothing is fetched and nothing is read from the workspace.
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 0, "{err}");
assert_eq!(w.fetches(), 0);
assert!(!w.0.join("rustup").exists());
// A stale workspace copy of the same version is never run.
std::fs::create_dir_all(w.tc("rustup").join("lib")).unwrap();
std::fs::write(w.tc("rustup").join("lib/librustc_driver.so"), "torn").unwrap();
let (code, out, err) = w.run(".", SYS_PATH, &[], "echo $RUSTUP_HOME");
assert_eq!((code, out.as_str()), (0, "/opt/rustup\n"), "{err}");
}
#[test]
fn provisioning_prunes_only_the_workspace_copy_served_by_the_template() {
let w = Ws::new("tc-image-prune");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
w.file(&format!("rustup/update-hashes/{TC}"), "old");
w.file("rustup/toolchains/other/bin/rustc", "keep");
w.file("rustup/update-hashes/other", "keep");
w.bake_image();
for _ in 0..2 {
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 0, "{err}");
assert!(!w.tc("rustup").exists());
assert!(!w.0.join("rustup/update-hashes").join(TC).exists());
assert!(w.0.join("rustup/toolchains/other/bin/rustc").exists());
assert!(w.0.join("rustup/update-hashes/other").exists());
assert!(w.tc("image-rustup").join("bin/rustc").exists());
assert_eq!(w.fetches(), 0);
}
}
#[test]
fn a_broken_template_toolchain_is_a_platform_failure() {
let w = Ws::new("tc-image-bad");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
w.bake_image();
std::fs::write(w.tc("image-rustup").join("lib/librustc_driver.so"), "").unwrap();
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 4, "{err}");
assert!(err.contains("cannot read file data"), "{err}");
assert_eq!(
Outcome::platform(provision_failure(code, err.as_bytes()), true).code(),
125
);
}
#[test]
fn a_corrupted_workspace_toolchain_is_fetched_again_never_run() {
let w = Ws::new("tc-corrupt");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
assert_eq!(w.fetches(), 1);
// One file torn on the Volume, the way fb 3718's driver was, and one
// left readable but altered: each is caught before the command.
for torn in ["", "driver\nextra\n"] {
std::fs::write(w.tc("rustup").join("lib/librustc_driver.so"), torn).unwrap();
let before = w.fetches();
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 0, "{err}");
assert_eq!(w.fetches(), before + 1);
let log = std::fs::read_to_string(w.0.join(".sylphx/toolchain.log")).unwrap();
assert!(log.contains("failed its check"), "{log}");
// The command runs the fresh copy.
let (code, out, err) = w.run(
".",
SYS_PATH,
&[("TC", TC)],
"\"$RUSTUP_HOME/toolchains/$TC/bin/rustc\" -vV",
);
assert_eq!((code, out.as_str()), (0, "rustc 1.99.0\n"), "{err}");
}
}
#[test]
fn a_component_added_by_a_command_keeps_the_workspace_toolchain() {
let w = Ws::new("tc-component");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
// What `rustup component add rust-src` changes and adds.
let lib = w.tc("rustup").join("lib/rustlib");
std::fs::write(lib.join("components"), "rustc\nrust-src\n").unwrap();
std::fs::write(lib.join("manifest-rust-src"), "file:lib/rustlib/src\n").unwrap();
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 0, "{err}");
assert_eq!(w.fetches(), 1);
}
#[test]
fn a_corrupted_workspace_toolchain_that_cannot_be_fetched_is_125() {
let w = Ws::new("tc-corrupt-offline");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
std::fs::write(w.tc("rustup").join("lib/librustc_driver.so"), "").unwrap();
w.file("offline", "");
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 4, "{err}");
assert_eq!(
Outcome::platform(provision_failure(code, err.as_bytes()), true).code(),
125
);
}
#[test]
fn a_half_written_or_unsealed_workspace_toolchain_is_installed_again() {
// Left by a lease that was lost mid-install: the directory is there,
// its rustc is not, so rustup alone would call it installed.
let w = Ws::new("tc-half");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
std::fs::create_dir_all(w.tc("rustup").join("lib")).unwrap();
let (code, err) = w.sh(PROVISION, ".", &[]);
assert_eq!(code, 0, "{err}");
assert!(w.tc("rustup").join("bin/rustc").exists());
// Installed by an older client, which wrote no checksums: once again.
std::fs::remove_file(w.tc("rustup").join(".sylphx-sha256")).unwrap();
let before = w.fetches();
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
assert_eq!(w.sh(PROVISION, ".", &[]).0, 0);
assert_eq!(w.fetches(), before + 1);
}
#[test]
fn toolchains_come_through_the_cache_tier_and_directly_when_it_fails() {
let dist = "http://build-cache.env-1.svc.cluster.local/upstream/static.rust-lang.org";
let cache: BTreeMap<String, String> = [
("RUSTUP_DIST_SERVER", dist.to_string()),
("RUSTUP_UPDATE_ROOT", format!("{dist}/rustup")),
("SCCACHE_WEBDAV_TOKEN", "a-run-token".to_string()),
(
"SYLPHX_CRATES_MIRROR",
"sparse+http://x/crates/index/".to_string(),
),
]
.into_iter()
.map(|(k, v)| (k.to_string(), v))
.collect();
// Only the toolchain tier's addresses; the run token stays out.
let env = provision_env(&cache);
assert_eq!(
env.keys().map(String::as_str).collect::<Vec<_>>(),
["RUSTUP_DIST_SERVER", "RUSTUP_UPDATE_ROOT"]
);
assert!(provision_env(&BTreeMap::new()).is_empty());
let pairs: Vec<(&str, &str)> = env.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
for (down, want) in [(false, dist), (true, "direct")] {
let w = Ws::new("tc-tier");
w.file("tree/Cargo.toml", "[workspace]\n");
w.rustup(false);
if down {
w.file("tier-down", "");
}
let (code, err) = w.sh_env(PROVISION, ".", &[], &pairs);
assert_eq!(code, 0, "{err}");
assert_eq!(
std::fs::read_to_string(w.0.join("fetches")).unwrap(),
format!("{want}\n")
);
}
}
#[test]
fn a_missing_tree_fails_provisioning() {
let w = Ws::new("no-tree");
let (code, err) = w.sh(PROVISION, "gone", &[]);
assert_eq!(code, 3, "{err}");
}
#[test]
fn the_commands_own_failure_keeps_its_exit_code() {
let w = Ws::new("run");
w.file("tree/x", "");
for want in [0, 1, 2, 7, 101] {
let (code, _) = w.sh(RUN, ".", &["0", "/bin/sh", "-c", &format!("exit {want}")]);
assert_eq!(code, want);
assert_eq!(Outcome::Ran(code).code() as i32, want);
}
// Not a platform failure: the code is the command's, not 125.
assert_ne!(Outcome::Ran(101).code(), EXIT_PLATFORM);
// The tree missing is a failure before the command starts.
assert_eq!(w.sh(RUN, "gone", &["0", "/bin/sh", "-c", "exit 0"]).0, 125);
}
#[test]
fn a_stream_error_is_retried_once_with_a_backoff_then_is_125() {
let mut r = Retries::default();
let stream =
|| Attempt::Lost("the process stream broke: error decoding response body".into());
match r.next(stream()) {
Step::Again { reason, backoff } => {
assert!(reason.contains("process stream broke"));
assert!(backoff >= Duration::from_secs(1), "{backoff:?}");
}
Step::Done(o) => panic!("the first loss is retried: {o:?}"),
}
match r.next(stream()) {
Step::Done(o) => {
assert_eq!(o.code(), 125);
assert!(
matches!(&o, Outcome::Platform { reason, retryable: true } if reason.ends_with(", twice"))
);
}
Step::Again { .. } => panic!("only one retry"),
}
}
#[test]
fn a_retry_that_succeeds_keeps_the_commands_exit_code() {
let mut r = Retries::default();
assert!(matches!(
r.next(Attempt::Lost("x".into())),
Step::Again { .. }
));
match r.next(Attempt::Done(Outcome::Ran(101))) {
Step::Done(o) => assert_eq!((o.code(), o), (101, Outcome::Ran(101))),
Step::Again { .. } => panic!("done is done"),
}
// A command that failed on its own is never retried.
match Retries::default().next(Attempt::Done(Outcome::Ran(1))) {
Step::Done(Outcome::Ran(1)) => {}
_ => panic!("a command's own failure is not retried"),
}
}
#[test]
fn exit_codes_follow_docker_run_and_timeout() {
assert_eq!(Outcome::Ran(0).code(), 0);
assert_eq!(
Outcome::Ran(101).code(),
101,
"the command's own status passes through"
);
assert_eq!(Outcome::Ran(2).code(), 2);
assert_eq!(Outcome::Ran(137).code(), 137);
assert_eq!(Outcome::Ran(256 + 3).code(), 3);
assert_eq!(Outcome::TimedOut.code(), 124);
assert_eq!(Outcome::platform("no capacity", true).code(), 125);
assert_eq!(Outcome::Interrupted.code(), 130);
assert_eq!(Outcome::Ran(0).name(), "succeeded");
assert_eq!(Outcome::Ran(1).name(), "failed");
}
#[test]
fn the_command_follows_the_double_dash() {
let tree = GitTree::new();
let o = parse(&[
"sylphx",
"build",
"run",
tree.path(),
"--",
"cargo",
"test",
"-p",
"x",
])
.unwrap();
assert_eq!(o.command, ["cargo", "test", "-p", "x"]);
assert_eq!(o.size, "large");
assert_eq!(o.timeout, DEFAULT_TIMEOUT);
assert_eq!(
o.region, None,
"no --region: automatic ephemeral Build placement"
);
assert_eq!(o.queue_timeout, DEFAULT_QUEUE_TIMEOUT);
assert!(!o.json);
let o = parse(&[
"sylphx",
"-o",
"json",
"build",
"run",
tree.path(),
"--size",
"xlarge",
"--timeout",
"90m",
"--env",
"RUST_LOG=debug",
"--allow-host",
"Proxy.Golang.org",
"--artifact",
"a/*",
"--",
"make",
])
.unwrap();
assert!(o.json);
assert_eq!(o.size, "xlarge");
assert_eq!(o.timeout, Duration::from_secs(5400));
assert_eq!(o.env.get("RUST_LOG").map(String::as_str), Some("debug"));
assert_eq!(o.allow_hosts, ["proxy.golang.org"]);
assert_eq!(o.artifacts, ["a/*"]);
}
/// A caller that prefers one region: a region and a short wait for a machine.
#[test]
fn region_and_queue_timeout_parse() {
let tree = GitTree::new();
let o = parse(&[
"sylphx",
"build",
"run",
tree.path(),
"--size",
"standard",
"--region",
"gra",
"--queue-timeout",
"120s",
"--timeout",
"60m",
"--",
"cargo",
"check",
])
.unwrap();
assert_eq!(o.region.as_deref(), Some("gra"));
assert_eq!(o.queue_timeout, Duration::from_secs(120));
assert_eq!(o.timeout, Duration::from_secs(3600));
}
/// A caller sends --region only when `build run --help` lists both
/// flags, so the help text is part of the contract.
#[test]
fn help_lists_region_and_queue_timeout() {
let help = command().render_long_help().to_string();
assert!(help.contains("--region"), "{help}");
assert!(help.contains("--queue-timeout"), "{help}");
assert!(help.contains("default 30m"), "{help}");
assert!(
help.contains("Example: sylphx build run --size standard -- cargo test"),
"{help}"
);
assert!(!help.contains("--queue-timeout 120s"), "{help}");
}
#[test]
fn the_default_queue_budget_covers_measured_cold_fallback() {
let tree = GitTree::new();
let o = parse(&["sylphx", "build", "run", tree.path(), "--", "true"]).unwrap();
assert_eq!(o.queue_timeout, Duration::from_secs(1800));
// The incident sample's nearest-rank p99 was 225.128s to READY;
// volume-pinned requeues made the longest caller wait over 450s.
assert!(o.queue_timeout > Duration::from_secs(455));
let explicit = parse(&[
"sylphx",
"build",
"run",
"--queue-timeout",
"120s",
tree.path(),
"--",
"true",
])
.unwrap();
assert_eq!(explicit.queue_timeout, Duration::from_secs(120));
}
#[test]
fn usage_errors_are_caught_before_anything_is_created() {
assert!(parse(&["sylphx", "build", "run"]).is_err(), "no command");
assert!(
parse(&["sylphx", "build", "run", "cargo"]).is_err(),
"no --"
);
for bad in [
&["--timeout", "7h"][..],
&["--timeout", "0s"],
&["--timeout", "soon"],
&["--queue-timeout", "0s"],
&["--queue-timeout", "7h"],
&["--queue-timeout", "later"],
&["--region", "GRA"],
&["--region", "gra west"],
&["--region", ""],
&["--region", "-gra"],
&["--size", "huge"],
&["--env", "NOVALUE"],
&["--env", "1X=y"],
&["--env", "SYLPHX_ENVIRONMENT=orgs/o/projects/p/envs/e"],
&["--allow-host", "a b"],
&["/nonexistent-dir-for-test"],
] {
let mut args = vec!["sylphx", "build", "run"];
args.extend_from_slice(bad);
args.extend_from_slice(&["--", "true"]);
assert!(parse(&args).is_err(), "{bad:?}");
}
}
#[test]
fn artifacts_never_land_outside_the_out_dir() {
let d = Path::new("/out");
assert_eq!(
artifact_path(d, "target/release/app"),
Some(PathBuf::from("/out/target/release/app"))
);
assert_eq!(
artifact_path(d, "../../target/x"),
Some(PathBuf::from("/out/target/x"))
);
assert_eq!(
artifact_path(d, "/etc/passwd"),
Some(PathBuf::from("/out/etc/passwd"))
);
assert_eq!(artifact_path(d, ".."), None);
}
#[test]
fn sha256sum_lines_parse() {
let h = "a".repeat(64);
assert_eq!(
parse_sum(&format!("{h} dir/file name")),
Some((h.as_str(), "dir/file name"))
);
assert_eq!(parse_sum(&format!("\\{h} odd\\nname")), None);
assert_eq!(parse_sum("short x"), None);
}
/// Every `-o json` event has a `type` the schema knows and that type's
/// required fields, with the right JSON kinds.
#[test]
fn json_events_match_the_schema() {
/// A required field and the JSON kind it must have.
type Field = (&'static str, fn(&Value) -> bool);
fn check(v: &Value) {
let t = v["type"].as_str().expect("type");
let need: &[Field] = match t {
"sync" => &[
("files", Value::is_u64),
("bytes", Value::is_u64),
("full", Value::is_boolean),
],
"queued" => &[],
"running" => &[("workspace", Value::is_string)],
"stdout" | "stderr" => &[("data", Value::is_string)],
"artifact" => &[
("path", Value::is_string),
("digest", Value::is_string),
("bytes", Value::is_u64),
],
"result" => &[
("exit_code", Value::is_u64),
("outcome", Value::is_string),
("retryable", Value::is_boolean),
("duration_ms", Value::is_u64),
("workspace", Value::is_string),
("bytes_up", Value::is_u64),
("bytes_down", Value::is_u64),
("from_store", Value::is_u64),
],
other => panic!("unknown event type {other}"),
};
for (k, ok) in need {
assert!(ok(&v[*k]), "{t}.{k} in {v}");
}
}
// The events as the code builds them.
let stats = Stats {
bytes_up: 512,
bytes_down: 2048,
artifacts: 0,
from_store: 7,
image: None,
};
let captured = Mutex::new(Vec::new());
let sink = |v: Value| captured.lock().unwrap().push(v);
for o in [
Outcome::Ran(0),
Outcome::Ran(101),
Outcome::TimedOut,
Outcome::Interrupted,
Outcome::platform("x", true),
] {
let r = result_event(&o, &stats, 1);
assert_eq!(r["exit_code"], json!(o.code()));
assert_eq!(r["retryable"], json!(o == Outcome::platform("x", true)));
sink(r);
}
sink(json!({"type": "stdout", "data": String::from_utf8_lossy(b"ok\n")}));
sink(
json!({"type": "sync", "full": true, "files": 1, "removed": 0, "bytes": 10, "bytes_up": 300, "duration_ms": 5, "workspace": "cold", "volume": false}),
);
sink(
json!({"type": "running", "lease": "l", "size": "large", "workspace": "cold", "volume": false}),
);
sink(json!({"type": "artifact", "path": "p", "digest": "sha256:aa", "bytes": 3}));
for v in captured.lock().unwrap().iter() {
check(v);
}
}
}