surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bound native cargo builds and individual WASI tests with cancellable deadlines

Browse Source
This commit is contained in:
surma
2026-10-25 20:14:32 +00:00
parent 895be183c9
commit c15e50d752
6 changed files with 533 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/cargo.rs
+205 -31
View File
@@ -4,11 +4,44 @@ use anyhow::{Context, Result, bail};
use include_dir::{Dir, include_dir};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::time::Duration;
use std::{
path::{Path, PathBuf},
sync::Arc,
};
use tokio::process::Command;
use tokio::{io::AsyncReadExt, process::Command, time::Instant};
/// Default wall-clock budget after acquiring the build mutex (not queue time).
pub const DEFAULT_TIMEOUT_SECS: u64 = 900;
/// Default wall-clock budget for each WASI test instance.
pub const DEFAULT_TEST_TIMEOUT_SECS: u64 = 120;
/// Wall-clock budgets for a build and its individual WASI tests.
#[derive(Clone, Copy, Debug)]
pub struct BuildOptions {
pub timeout_secs: u64,
pub test_timeout_secs: u64,
}
impl Default for BuildOptions {
fn default() -> Self {
Self {
timeout_secs: DEFAULT_TIMEOUT_SECS,
test_timeout_secs: DEFAULT_TEST_TIMEOUT_SECS,
}
}
}
/// Optional RPC seconds are positive integers; null does not disable a deadline.
fn seconds(params: &Value, name: &str, default: u64) -> Result<u64, Value> {
match params.get(name) {
None => Ok(default),
Some(value) => value
.as_u64()
.filter(|seconds| *seconds >= 1)
.ok_or_else(|| {
crate::error(-32602, &format!("{name} must be an integer of at least 1"))
}),
}
}
/// The native build plugin.
pub const ID: &str = "cargo";
@@ -38,6 +71,8 @@ pub struct BuildFailure {
pub reason: String,
pub tests: TestResults,
pub output: String,
pub elapsed_secs: Option<f64>,
pub step: Option<&'static str>,
}
impl std::fmt::Display for BuildFailure {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -51,6 +86,8 @@ impl From<anyhow::Error> for BuildFailure {
reason: "build preparation failed".into(),
output: format!("{error:#}"),
tests: TestResults::default(),
elapsed_secs: None,
step: None,
}
}
}
@@ -142,19 +179,39 @@ impl Cargo {
host.register_native(ID,Arc::new(move |_context,method,params| {
let builder=builder.clone();
Box::pin(async move {
if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","params":[{"name":"plugin","required":true,"schema":{"type":"string"}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings"],"properties":{"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); }
if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","description":"Copies source, runs bounded WASI tests, then builds a release component. Deadlines kill cargo and its descendants; cancellation releases the build mutex.","params":[{"name":"plugin","required":true,"schema":{"type":"string"}},{"name":"timeout_secs","required":false,"description":"Total wall-clock limit in seconds for tests and release build after acquiring the build mutex (queue time excluded). Default 900; minimum 1. Cannot be disabled.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TIMEOUT_SECS}},{"name":"test_timeout_secs","required":false,"description":"Wall-clock limit in seconds for each WASI test instance. Default 120; minimum 1. A timed-out test is named and fails; remaining tests continue.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TEST_TIMEOUT_SECS}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings"],"properties":{"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); }
if method!="build" { return Err(crate::error(-32601,"cargo method not found")); }
let plugin=params["plugin"].as_str().ok_or_else(||crate::error(-32602,"plugin must be a string"))?;
match builder.build(&builder.world.plugin_dir(plugin).join("source")).await {
let options = BuildOptions {
timeout_secs: seconds(&params, "timeout_secs", DEFAULT_TIMEOUT_SECS)?,
test_timeout_secs: seconds(&params, "test_timeout_secs", DEFAULT_TEST_TIMEOUT_SECS)?,
};
match builder.build_with_options(&builder.world.plugin_dir(plugin).join("source"), options).await {
Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings})),
Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"tests":failure.tests,"output":failure.output}})),
Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"reason":failure.reason,"tests":failure.tests,"output":failure.output,"elapsed_secs":failure.elapsed_secs,"step":failure.step}})),
}
})
}));
}
/// Copies source once, runs WASI tests, then publishes a release component under state/cargo/builds.
pub async fn build(&self, source: &Path) -> std::result::Result<BuildOutput, BuildFailure> {
self.build_with_options(source, BuildOptions::default())
.await
}
/// Like build, with explicit positive wall-clock budgets. Waiting for the mutex is unbounded.
pub async fn build_with_options(
&self,
source: &Path,
options: BuildOptions,
) -> std::result::Result<BuildOutput, BuildFailure> {
if options.timeout_secs == 0 || options.test_timeout_secs == 0 {
return Err(anyhow::anyhow!("timeouts must be at least 1 second").into());
}
let build_guard = self.builds.clone().lock_owned().await;
let started = Instant::now();
let deadline = started
.checked_add(Duration::from_secs(options.timeout_secs))
.ok_or_else(|| anyhow::anyhow!("timeout_secs is too large"))?;
let work = self.world.state_dir(ID).join("work");
let target = self.target.clone();
let temporary = self.world.state_dir(ID).join("tmp");
@@ -208,16 +265,27 @@ impl Cargo {
.await
.map_err(anyhow::Error::from)??;
let mut test = Command::new("cargo");
test.args(["test", "--release", "--target", TARGET, "--target-dir"])
.arg(&target)
.current_dir(&work)
.env("TMPDIR", &temporary)
.arg("--config")
.arg(format!(
"target.wasm32-wasip2.runner={}",
json!([self.runner, "test-runner"])
));
let output = run_command(test).await?;
test.args([
"test",
"--no-fail-fast",
"--release",
"--target",
TARGET,
"--target-dir",
])
.arg(&target)
.current_dir(&work)
.env("TMPDIR", &temporary)
.env(
"SAUSAGE_TEST_TIMEOUT_SECS",
options.test_timeout_secs.to_string(),
)
.arg("--config")
.arg(format!(
"target.wasm32-wasip2.runner={}",
json!([self.runner, "test-runner"])
));
let output = run_command(test, started, deadline, "tests", TestResults::default()).await?;
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
let tests = TestResults {
@@ -234,22 +302,24 @@ impl Cargo {
reason: "WASI tests failed".into(),
output: tests.output.clone(),
tests,
elapsed_secs: None,
step: None,
});
}
let mut build = Command::new("cargo");
build
.args([
"build",
"--release",
"--target",
TARGET,
"--message-format=json-render-diagnostics",
"--target-dir",
])
.arg(&target)
.current_dir(&work);
let output = run_command(build, started, deadline, "release build", tests.clone()).await?;
let result = async {
let mut build = Command::new("cargo");
build
.args([
"build",
"--release",
"--target",
TARGET,
"--message-format=json-render-diagnostics",
"--target-dir",
])
.arg(&target)
.current_dir(&work);
let output = run_command(build).await?;
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
let diagnostics = stderr.to_string();
@@ -292,6 +362,8 @@ impl Cargo {
reason: "release build failed".into(),
output: format!("{error:#}"),
tests,
elapsed_secs: None,
step: None,
})
}
}
@@ -306,16 +378,81 @@ impl Drop for ProcessGroup {
}
}
/// Starts one process group and retains its guard across all output collection.
async fn run_command(mut command: Command) -> Result<std::process::Output> {
async fn run_command(
mut command: Command,
started: Instant,
deadline: Instant,
step: &'static str,
mut tests: TestResults,
) -> std::result::Result<std::process::Output, BuildFailure> {
command
.process_group(0)
.kill_on_drop(true)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped());
let child = command.spawn().context("start cargo")?;
let _group = ProcessGroup(child.id().unwrap());
Ok(child.wait_with_output().await?)
let mut child = command.spawn().context("start cargo")?;
let group = ProcessGroup(child.id().unwrap());
let mut stdout = child.stdout.take().unwrap();
let mut stderr = child.stderr.take().unwrap();
let mut out = Vec::new();
let mut err = Vec::new();
let (status, timed_out) = {
let collect = async {
let (status, _, _) = tokio::try_join!(
child.wait(),
stdout.read_to_end(&mut out),
stderr.read_to_end(&mut err)
)?;
Ok::<_, std::io::Error>(status)
};
tokio::pin!(collect);
tokio::select! {
biased;
_ = tokio::time::sleep_until(deadline) => {
// Kill the entire tree before waiting for EOF/reaping cargo. The guard also
// covers cancellation while collecting, and runs before the mutex drops.
let _ = nix::sys::signal::killpg(nix::unistd::Pid::from_raw(group.0 as i32), nix::sys::signal::Signal::SIGKILL);
(collect.await, true)
}
status = &mut collect => (status, false),
}
};
let status = status.map_err(anyhow::Error::from)?;
if timed_out {
let output = format!(
"{}\n{}",
String::from_utf8_lossy(&out),
String::from_utf8_lossy(&err)
);
let output = output
.trim_end()
.lines()
.rev()
.take(50)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect::<Vec<_>>()
.join("\n");
if step == "tests" {
tests.passed = test_count(&String::from_utf8_lossy(&out), "passed");
tests.failed = test_count(&String::from_utf8_lossy(&out), "failed");
tests.output = output.clone();
}
return Err(BuildFailure {
reason: "timeout".into(),
tests,
output,
elapsed_secs: Some(started.elapsed().as_secs_f64()),
step: Some(step),
});
}
Ok(std::process::Output {
status,
stdout: out,
stderr: err,
})
}
/// Copies only source, excluding build outputs and git internals at every level.
fn copy_source(source: &Path, destination: &Path, root: bool) -> Result<()> {
@@ -457,3 +594,40 @@ fn strip_progress(output: &str) -> String {
.collect::<Vec<_>>()
.join("\n")
}
#[cfg(test)]
mod timeout_tests {
use super::*;
/// Deadline errors retain the shared start time, phase, test totals, and output tail.
#[tokio::test]
async fn timeout_diagnostics() {
let mut command = Command::new("sh");
command.args([
"-c",
"i=0; while [ $i -lt 100 ]; do echo diagnostic-$i; i=$((i+1)); done; exec sleep 1000",
]);
let now = Instant::now();
let failure = run_command(
command,
now - Duration::from_secs(10),
now + Duration::from_secs(1),
"release build",
TestResults {
passed: 7,
failed: 0,
output: "successful tests".into(),
},
)
.await
.unwrap_err();
assert_eq!(failure.reason, "timeout");
assert_eq!(failure.step, Some("release build"));
assert!(failure.elapsed_secs.unwrap() >= 11.0);
assert_eq!(failure.tests.passed, 7);
assert_eq!(failure.tests.output, "successful tests");
assert_eq!(failure.output.lines().count(), 50);
assert!(failure.output.contains("diagnostic-99"));
assert!(!failure.output.contains("diagnostic-0\n"));
}
}
cli/src/test_runner.rs
+19 -2
View File
@@ -64,12 +64,21 @@ async fn instance(
("SAUSAGE_PLUGIN".into(), "test".into()),
("SAUSAGE_PARENT".into(), "core".into()),
]);
let timeout_secs = std::env::var("SAUSAGE_TEST_TIMEOUT_SECS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
.filter(|value| *value >= 1)
.unwrap_or(sausage::cargo::DEFAULT_TEST_TIMEOUT_SECS);
let deadline = tokio::time::Instant::now()
.checked_add(std::time::Duration::from_secs(timeout_secs))
.ok_or_else(|| anyhow::anyhow!("test_timeout_secs is too large"))?;
let (ended, mut ended_rx) = tokio::sync::watch::channel(true);
let mut guest = runtime
.start_with_args(
&component,
InstanceConfig {
filesystem,
store_ended: tokio::sync::watch::channel(true).0,
store_ended: ended,
network: true,
env,
},
@@ -88,6 +97,7 @@ async fn instance(
let mut stdout_open = true;
let mut stderr_open = true;
let mut handlers = tokio::task::JoinSet::new();
let result = tokio::time::timeout_at(deadline, async {
while stdout_open || stderr_open || !pending.is_empty() {
tokio::select! {
line=async { if let Some(line) = pending.pop_front() { Some(line) } else { guest.stdout.recv().await } }, if stdout_open || !pending.is_empty()=>match line {
@@ -129,11 +139,18 @@ async fn instance(
},
}
}
Ok::<_, anyhow::Error>(guest.wait().await)
}).await;
handlers.abort_all();
let exit = guest.wait().await;
// Dropping the instance aborts even a CPU-bound WASI loop. Wait for the
// store and outstanding filesystem work before removing its test world.
drop(guest);
let _ = ended_rx.wait_for(|ended| *ended).await;
let _ = stderr_task.await;
tracker.idle().await;
host.stop().await;
let exit =
result.map_err(|_| anyhow::anyhow!("timed out after {timeout_secs} s\n{output}"))??;
Ok(Run { exit, output })
}
/// Extracts libtest names from its stable list format.