//! Native cargo builds, one immutable source copy, SDK installation, and process-group cancellation. use crate::{Host, world::World}; use anyhow::{Context, Result, bail}; use include_dir::{Dir, include_dir}; use serde_json::{Value, json}; use std::time::Duration; use std::{ path::{Path, PathBuf}, sync::Arc, }; use tokio::{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 { 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"; /// The target used for builds and tests. pub const TARGET: &str = "wasm32-wasip2"; static SDK: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk/src"); static MACROS: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk/macros/src"); static INTERFACE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../interface/src"); #[derive(Clone, Default, Debug, serde::Serialize)] /// Test totals and captured output. pub struct TestResults { pub passed: u64, pub failed: u64, pub output: String, } /// Raw, unfiltered output shared by the test and release phases of one build. #[derive(Clone, Debug, serde::Serialize)] pub struct BuildLogs { pub stdout: StreamLog, pub stderr: StreamLog, pub total_bytes: u64, } #[derive(Clone, Debug, serde::Serialize)] pub struct StreamLog { pub path: String, pub bytes: u64, } /// Persistent descriptors are inherited by cargo, so cancellation cannot lose /// bytes already written by cargo or its descendants. No buffered writer is used. struct LogFiles { stdout: std::fs::File, stderr: std::fs::File, stdout_path: PathBuf, stderr_path: PathBuf, } impl LogFiles { fn create(world: &Path) -> Result { let directory = world .join("logs/cargo") .join(chrono::Utc::now().format("%Y-%m-%d").to_string()); std::fs::create_dir_all(&directory).context("create cargo log directory")?; let id = uuid::Uuid::now_v7(); let stdout_path = directory.join(format!("{id}.stdout.log")); let stderr_path = directory.join(format!("{id}.stderr.log")); let create = |path: &Path| { std::fs::OpenOptions::new() .create_new(true) .append(true) .open(path) .with_context(|| format!("create {}", path.display())) }; Ok(Self { stdout: create(&stdout_path)?, stderr: create(&stderr_path)?, stdout_path, stderr_path, }) } fn info(&self, world: &Path) -> Result { let stream = |file: &std::fs::File, path: &Path| -> Result { Ok(StreamLog { path: format!("/{}", path.strip_prefix(world)?.display()), bytes: file.metadata()?.len(), }) }; let stdout = stream(&self.stdout, &self.stdout_path)?; let stderr = stream(&self.stderr, &self.stderr_path)?; Ok(BuildLogs { total_bytes: stdout.bytes + stderr.bytes, stdout, stderr, }) } fn read(&self, path: &Path, offset: u64) -> Result> { use std::io::{Read, Seek, SeekFrom}; let mut file = std::fs::File::open(path).with_context(|| format!("read {}", path.display()))?; file.seek(SeekFrom::Start(offset))?; let mut bytes = Vec::new(); file.read_to_end(&mut bytes)?; Ok(bytes) } } /// A successful build, usable without a running host. #[derive(Debug)] pub struct BuildOutput { pub logs: Option, pub wasm: PathBuf, pub warnings: String, pub tests: TestResults, } /// A failed compiler or test phase, with its complete diagnostics. #[derive(Debug)] pub struct BuildFailure { pub logs: Option, pub reason: String, pub tests: TestResults, pub output: String, pub elapsed_secs: Option, pub step: Option<&'static str>, } impl std::fmt::Display for BuildFailure { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(formatter, "{}\n{}", self.reason, self.output) } } impl std::error::Error for BuildFailure {} impl From for BuildFailure { fn from(error: anyhow::Error) -> Self { Self { logs: None, reason: "build preparation failed".into(), output: format!("{error:#}"), tests: TestResults::default(), elapsed_secs: None, step: None, } } } /// A host-independent builder. Each builder serializes access to its work and target folders. pub struct Cargo { world: World, builds: Arc>, runner: PathBuf, target: PathBuf, sdk: PathBuf, } impl Cargo { /// Seeds a missing world SDK and selects the current sausage executable as runner. pub fn new(world: World, cache: &Path) -> Result { let mut runner = std::env::current_exe()?; if runner .parent() .is_some_and(|parent| parent.file_name().is_some_and(|name| name == "deps")) { runner = runner.parent().unwrap().parent().unwrap().join("sausage"); } Self::with_runner(world, runner, cache) } /// Seeds a missing world SDK and accepts an explicit sausage binary, including for init integrations. pub fn with_runner(world: World, runner: PathBuf, cache: &Path) -> Result { let sdk = world.plugin_dir(ID).join("sdk"); Self::seed_sdk(&sdk)?; Ok(Self { sdk, target: std::path::absolute(cache)?.join("cargo-target"), world, runner, builds: Arc::new(tokio::sync::Mutex::new(())), }) } /// Seeds only an absent SDK; an existing world owns all of its SDK files. fn seed_sdk(sdk: &Path) -> Result<()> { if sdk.try_exists()? { return Ok(()); } let mut files = std::collections::BTreeMap::new(); collect_files(&SDK, Path::new("src"), &mut files); collect_files(&MACROS, Path::new("macros/src"), &mut files); collect_files(&INTERFACE, Path::new("interface/src"), &mut files); for (manifest, destination, interface_path) in [ ( include_str!("../../sdk/Cargo.toml"), PathBuf::from("Cargo.toml"), Some("interface"), ), ( include_str!("../../sdk/macros/Cargo.toml"), PathBuf::from("macros/Cargo.toml"), Some("../interface"), ), ( include_str!("../../interface/Cargo.toml"), PathBuf::from("interface/Cargo.toml"), None, ), ] { let mut document: toml_edit::DocumentMut = manifest.parse()?; if !document.contains_key("workspace") { document["workspace"] = toml_edit::table(); } if let Some(path) = interface_path { for section in ["dependencies", "dev-dependencies"] { if let Some(dependency) = document .get_mut(section) .and_then(|table| table.get_mut("sausage-interface")) { dependency["path"] = toml_edit::value(path); } } } files.insert(destination, document.to_string().into_bytes()); } let parent = sdk.parent().context("SDK has no parent")?; std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; let staging = tempfile::Builder::new() .prefix("seed-sdk-") .tempdir_in(parent) .with_context(|| format!("stage SDK in {}", parent.display()))?; for (path, bytes) in &files { write_changed(&staging.path().join(path), bytes)?; } // Another builder may have seeded the world while we prepared the files. if !sdk.try_exists()? && let Err(error) = std::fs::rename(staging.path(), sdk) && !sdk.try_exists()? { return Err(error).with_context(|| format!("seed {}", sdk.display())); } Ok(()) } /// Registers cargo.build as a cancellable native process. pub fn register(self, host: &Host) { let builder = Arc::new(self); 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","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","logs"],"properties":{"logs":{"type":"object","required":["stdout","stderr","total_bytes"],"properties":{"stdout":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"stderr":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"total_bytes":{"type":"integer","minimum":0}}},"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"))?; let options = BuildOptions { timeout_secs: seconds(¶ms, "timeout_secs", DEFAULT_TIMEOUT_SECS)?, test_timeout_secs: seconds(¶ms, "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,"logs":output.logs})), 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,"logs":failure.logs}})), } }) })); } /// 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 { 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 { let logs = LogFiles::create(&self.world.root)?; let result = self.build_logged(source, options, &logs).await; let info = logs.info(&self.world.root)?; match result { Ok(mut output) => { output.logs = Some(info); Ok(output) } Err(mut failure) => { failure.logs = Some(info); Err(failure) } } } async fn build_logged( &self, source: &Path, options: BuildOptions, logs: &LogFiles, ) -> std::result::Result { 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"); let cleanup = temporary.clone(); let source = source.to_path_buf(); let copy = work.clone(); let sdk = self.sdk.clone(); let build_guard = tokio::task::spawn_blocking(move || -> Result<_> { Self::seed_sdk(&sdk)?; if cleanup.exists() { std::fs::remove_dir_all(&cleanup) .with_context(|| format!("remove {}", cleanup.display()))?; } std::fs::create_dir_all(&cleanup) .with_context(|| format!("create {}", cleanup.display()))?; if copy.exists() { std::fs::remove_dir_all(©) .with_context(|| format!("remove {}", copy.display()))?; } copy_source(&source, ©, true)?; let manifest = copy.join("Cargo.toml"); let text = std::fs::read_to_string(&manifest) .with_context(|| format!("read {}", manifest.display()))?; std::fs::write( &manifest, prepare_manifest(&text, &sdk) .with_context(|| format!("prepare {}", manifest.display()))?, ) .with_context(|| format!("write {}", manifest.display()))?; Ok(build_guard) }) .await .map_err(anyhow::Error::from)??; let mut test = Command::new("cargo"); 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(), logs, ) .await?; let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); let tests = TestResults { passed: test_count(&stdout, "passed"), failed: if output.status.success() { 0 } else { test_count(&stdout, "failed").max(1) }, output: strip_progress(&format!("{stdout}\n{stderr}")), }; if !output.status.success() { return Err(BuildFailure { logs: None, 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(), logs, ) .await?; let result = async { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); let diagnostics = stderr.to_string(); let mut artifacts = Vec::new(); for line in stdout.lines() { if let Ok(message) = serde_json::from_str::(line) && message["reason"] == "compiler-artifact" && message["target"]["kind"] .as_array() .is_some_and(|kinds| kinds.iter().any(|kind| kind == "bin")) && let Some(executable) = message["executable"].as_str() { artifacts.push(PathBuf::from(executable)); } } if !output.status.success() { bail!("{diagnostics}"); } if artifacts.len() != 1 { bail!( "cargo must build exactly one binary, found {}", artifacts.len() ); } let artifact = artifacts.pop().unwrap(); let builds = self.world.state_dir(ID).join("builds"); let destination = tokio::task::spawn_blocking(move || { let _guard = build_guard; publish(&builds, &artifact) }) .await??; Ok(BuildOutput { logs: None, wasm: destination, warnings: strip_progress(&diagnostics), tests: tests.clone(), }) } .await; result.map_err(|error: anyhow::Error| BuildFailure { logs: None, reason: "release build failed".into(), output: format!("{error:#}"), tests, elapsed_secs: None, step: None, }) } } /// Kills cargo and every descendant, including tests and build scripts, on cancellation. struct ProcessGroup(u32); impl Drop for ProcessGroup { fn drop(&mut self) { let _ = nix::sys::signal::killpg( nix::unistd::Pid::from_raw(self.0 as i32), nix::sys::signal::Signal::SIGKILL, ); } } /// Starts one process group and retains its guard across all output collection. async fn run_command( mut command: Command, started: Instant, deadline: Instant, step: &'static str, mut tests: TestResults, logs: &LogFiles, ) -> std::result::Result { command .process_group(0) .kill_on_drop(true) .stdin(std::process::Stdio::null()) // Child writes go straight to persistent files: no pipe buffers or reader // tasks can be dropped on cancellation. Parsing happens only afterwards. .stdout(logs.stdout.try_clone().context("clone stdout log")?) .stderr(logs.stderr.try_clone().context("clone stderr log")?); let out_start = logs.stdout.metadata().context("stat stdout log")?.len(); let err_start = logs.stderr.metadata().context("stat stderr log")?.len(); let mut child = command.spawn().context("start cargo")?; let group = ProcessGroup(child.id().unwrap()); let (status, timed_out) = { let collect = child.wait(); 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)?; // Also stop any descendants that kept the streams open after cargo exited. drop(group); let out = logs.read(&logs.stdout_path, out_start)?; let err = logs.read(&logs.stderr_path, err_start)?; 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::>() .into_iter() .rev() .collect::>() .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 { logs: None, 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<()> { std::fs::create_dir_all(destination) .with_context(|| format!("create {}", destination.display()))?; for entry in std::fs::read_dir(source).with_context(|| format!("read {}", source.display()))? { let entry = entry.with_context(|| format!("read entry in {}", source.display()))?; if root && matches!(entry.file_name().to_str(), Some("target" | ".git")) { continue; } let from = entry.path(); let to = destination.join(entry.file_name()); if from.is_dir() { copy_source(&from, &to, false)?; } else { std::fs::copy(&from, &to) .with_context(|| format!("copy {} to {}", from.display(), to.display()))?; } } Ok(()) } /// Repoints SDK dependencies without discarding feature flags and detaches the copied crate's workspace. fn prepare_manifest(text: &str, sdk: &Path) -> Result { let mut document: toml_edit::DocumentMut = text.parse()?; rewrite_dependencies(document.as_table_mut(), sdk); if !document.contains_key("workspace") { document["workspace"] = toml_edit::table(); } Ok(document.to_string()) } /// Rewrites dependency entries recursively, including target and workspace tables. fn rewrite_dependencies(table: &mut dyn toml_edit::TableLike, sdk: &Path) { for (name, item) in table.iter_mut() { if matches!( &*name, "dependencies" | "dev-dependencies" | "build-dependencies" ) { if let Some(dependencies) = item.as_table_like_mut() { for (name, dependency) in dependencies.iter_mut() { if dependency .get("workspace") .and_then(toml_edit::Item::as_bool) == Some(true) { continue; } if &*name == "sausage-plugin" || dependency.get("package").and_then(toml_edit::Item::as_str) == Some("sausage-plugin") { if dependency.is_str() { let version = dependency.as_str().unwrap().to_string(); *dependency = toml_edit::value(toml_edit::InlineTable::new()); dependency["version"] = toml_edit::value(version); } dependency["path"] = toml_edit::value(sdk.display().to_string()); } } } } else if let Some(nested) = item.as_table_like_mut() { rewrite_dependencies(nested, sdk); } } } /// Publishes a complete artifact and retains only the ten newest numeric build folders. fn publish(builds: &Path, artifact: &Path) -> Result { std::fs::create_dir_all(builds).with_context(|| format!("create {}", builds.display()))?; let mut numbers: Vec = std::fs::read_dir(builds) .with_context(|| format!("read {}", builds.display()))? .filter_map(|entry| entry.ok()?.file_name().to_str()?.parse().ok()) .collect(); numbers.sort_unstable(); let number = numbers.last().copied().unwrap_or(0) + 1; let destination = builds .join(number.to_string()) .join(artifact.file_name().unwrap()); crate::world::write_atomic( &destination, &std::fs::read(artifact).with_context(|| format!("read {}", artifact.display()))?, ) .with_context(|| format!("publish {}", destination.display()))?; for old in numbers.iter().take(numbers.len().saturating_sub(9)) { let path = builds.join(old.to_string()); std::fs::remove_dir_all(&path).with_context(|| format!("remove {}", path.display()))?; } Ok(destination) } /// Writes embedded sources only when their bytes changed. fn write_changed(path: &Path, bytes: &[u8]) -> Result<()> { if std::fs::read(path).ok().as_deref() != Some(bytes) { std::fs::create_dir_all(path.parent().unwrap()) .with_context(|| format!("create parent of {}", path.display()))?; std::fs::write(path, bytes).with_context(|| format!("write {}", path.display()))?; } Ok(()) } /// Collects embedded source files for seeding a missing world SDK. fn collect_files( directory: &Dir<'_>, destination: &Path, files: &mut std::collections::BTreeMap>, ) { for file in directory.files() { files.insert(destination.join(file.path()), file.contents().to_vec()); } for directory in directory.dirs() { collect_files(directory, destination, files); } } /// Counts only final libtest summaries, never individual status lines. fn test_count(output: &str, field: &str) -> u64 { output .lines() .filter_map(|line| line.strip_prefix("test result:")) .flat_map(|summary| summary.split(';')) .filter_map(|part| { let words: Vec<_> = part.split_whitespace().collect(); let index = words.iter().position(|word| *word == field)?; words.get(index.checked_sub(1)?)?.parse::().ok() }) .sum() } /// Removes progress lines while retaining compiler diagnostics. fn strip_progress(output: &str) -> String { output .lines() .filter(|line| { ![ "Compiling ", "Finished ", "Blocking ", "Updating ", "Locking ", "Adding ", ] .iter() .any(|prefix| line.trim_start().starts_with(prefix)) }) .collect::>() .join("\n") } #[cfg(test)] mod timeout_tests { use super::*; /// Streams retain exact bytes, including non-UTF8, progress and JSON; phases append. #[tokio::test] async fn raw_streams_append_before_parsing() { let directory = tempfile::tempdir().unwrap(); let logs = LogFiles::create(directory.path()).unwrap(); for _ in 0..2 { let mut command = Command::new("sh"); command.args(["-c", "printf '\\377raw\\r'; printf ' Compiling progress\\n{\"reason\":\"compiler-artifact\"}\\n' >&2; exit 3"]); let now = Instant::now(); let output = tokio::time::timeout( Duration::from_secs(5), run_command( command, now, now + Duration::from_secs(3), "tests", TestResults::default(), &logs, ), ) .await .unwrap() .unwrap(); assert!(!output.status.success()); assert_eq!(output.stdout, b"\xffraw\r"); assert!(String::from_utf8_lossy(&output.stderr).contains("Compiling progress")); } assert_eq!( std::fs::read(&logs.stdout_path).unwrap(), b"\xffraw\r\xffraw\r" ); let info = logs.info(directory.path()).unwrap(); assert_eq!(info.stdout.bytes, 10); assert_eq!(info.total_bytes, info.stdout.bytes + info.stderr.bytes); assert!(info.stdout.path.starts_with("/logs/cargo/")); assert!(!directory.path().join("state/cargo").exists()); } /// Cancellation leaves kernel-written data intact, even without a newline. #[tokio::test] async fn cancelled_command_keeps_logs() { use notify::Watcher; let directory = tempfile::tempdir().unwrap(); let logs = LogFiles::create(directory.path()).unwrap(); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); let mut watcher = notify::recommended_watcher(move |event: notify::Result| { if let Ok(event) = event { if event .paths .iter() .any(|path| path.file_name().is_some_and(|name| name == "ready")) { let _ = tx.send(()); } } }) .unwrap(); watcher .watch(directory.path(), notify::RecursiveMode::NonRecursive) .unwrap(); let mut command = Command::new("sh"); command.args(["-c", "printf 'partial\\377'; printf 'stderr\\r' >&2; touch \"$1/ready\"; exec sleep 1000", "sh"]).arg(directory.path()); let now = Instant::now(); let task = tokio::spawn(async move { run_command( command, now, now + Duration::from_secs(60), "tests", TestResults::default(), &logs, ) .await }); tokio::time::timeout(Duration::from_secs(5), rx.recv()) .await .unwrap() .unwrap(); task.abort(); assert!(task.await.unwrap_err().is_cancelled()); let paths: Vec<_> = std::fs::read_dir(directory.path().join("logs/cargo")) .unwrap() .flat_map(|day| std::fs::read_dir(day.unwrap().path()).unwrap()) .map(|file| file.unwrap().path()) .collect(); assert_eq!(paths.len(), 2); for path in paths { let expected: &[u8] = if path.to_string_lossy().ends_with(".stdout.log") { b"partial\xff" } else { b"stderr\r" }; assert_eq!(std::fs::read(path).unwrap(), expected); } } /// 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 directory = tempfile::tempdir().unwrap(); let logs = LogFiles::create(directory.path()).unwrap(); 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(), }, &logs, ) .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")); let raw = std::fs::read_to_string(&logs.stdout_path).unwrap(); assert_eq!(raw.lines().count(), 100); assert!(raw.starts_with("diagnostic-0\n")); } }