diff --git a/crates/evix-cli/tests/eval.rs b/crates/evix-cli/tests/eval.rs index dbbef3a..8ade9e2 100644 --- a/crates/evix-cli/tests/eval.rs +++ b/crates/evix-cli/tests/eval.rs @@ -1,8 +1,9 @@ use std::{ + io::Read as _, net::{TcpListener, TcpStream}, - process::{Child, Command, Stdio}, + process::{Child, Command, Output, Stdio}, thread, - time::Duration, + time::{Duration, Instant}, }; fn evix() -> Command { @@ -75,6 +76,107 @@ fn remote_worker_consumes_shared_eval_queue() { assert!(stdout.contains(r#""name":"evix-remote""#), "{stdout}"); } +#[test] +fn derivation_outputs_carry_store_paths() { + let output = evix() + .args([ + "eval", + "--no-daemon", + "--expr", + "let system = builtins.currentSystem; in { recurseForDerivations = \ + true; pkg = derivation { name = \"evix-outputs\"; inherit system; \ + builder = \"/bin/sh\"; args = [ \"-c\" \"echo ok > $out\" ]; outputs = \ + [ \"out\" \"dev\" ]; }; }", + ]) + .output() + .expect("run evix"); + + assert!( + output.status.success(), + "status: {}\nstderr:\n{}", + output.status, + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + let line = stdout + .lines() + .find(|line| line.contains(r#""drvPath""#)) + .unwrap_or_else(|| panic!("no derivation event\n{stdout}")); + let event: serde_json::Value = + serde_json::from_str(line).expect("parse derivation event"); + for name in ["out", "dev"] { + let path = event["outputs"][name].as_str(); + assert!( + path.is_some_and(|path| path.starts_with("/nix/store/")), + "output {name} is {path:?}\n{stdout}" + ); + } +} + +#[test] +fn cyclic_attrsets_stop_at_the_traversal_depth_limit() { + let output = run_with_timeout( + evix().args([ + "eval", + "--no-daemon", + "--expr", + "let a = { recurseForDerivations = true; loop = a; }; in a", + ]), + Duration::from_secs(60), + ); + + assert!( + output.status.success(), + "status: {}\nstderr:\n{}", + output.status, + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.contains("maximum traversal depth"), "{stdout}"); +} + +/// Run `command` to completion, killing it once `limit` elapses. The pipes +/// drain on threads because a full one blocks the child and looks like a hang. +fn run_with_timeout(command: &mut Command, limit: Duration) -> Output { + let mut child = command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn evix"); + let mut child_stdout = child.stdout.take().expect("evix stdout"); + let mut child_stderr = child.stderr.take().expect("evix stderr"); + let stdout = thread::spawn(move || { + let mut buf = Vec::new(); + let _ = child_stdout.read_to_end(&mut buf); + buf + }); + let stderr = thread::spawn(move || { + let mut buf = Vec::new(); + let _ = child_stderr.read_to_end(&mut buf); + buf + }); + + let deadline = Instant::now() + limit; + let status = loop { + match child.try_wait().expect("poll evix") { + Some(status) => break status, + None if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + panic!("evix did not terminate within {limit:?}"); + }, + None => thread::sleep(Duration::from_millis(50)), + } + }; + + Output { + status, + stdout: stdout.join().expect("drain evix stdout"), + stderr: stderr.join().expect("drain evix stderr"), + } +} + fn unused_loopback_endpoint() -> String { let listener = TcpListener::bind("127.0.0.1:0").expect("bind test port"); let addr = listener.local_addr().expect("read test port"); diff --git a/crates/evix/src/eval.rs b/crates/evix/src/eval.rs index 1f366d7..6f29065 100644 --- a/crates/evix/src/eval.rs +++ b/crates/evix/src/eval.rs @@ -15,6 +15,9 @@ use crate::{EvalError, Event}; const NIX_GCROOTS_DIR: &str = "/nix/var/nix/gcroots"; +/// Attribute-path length at which traversal stops descending. +const MAX_ATTR_DEPTH: usize = 64; + #[derive(Debug, Clone)] pub(crate) struct EvalOptions { pub(crate) force_recurse: bool, @@ -75,6 +78,16 @@ pub fn process_attr<'s>( }, } }, + Ok(None) if path.len() >= MAX_ATTR_DEPTH => { + Event::Error(EvalError { + attr, + attr_path: path.to_vec(), + error: format!( + "attrset exceeds the maximum traversal depth of {MAX_ATTR_DEPTH}" + ), + fatal: false, + }) + }, Ok(None) => { let children = collect_recurse(&value, path, options.force_recurse); Event::AttrSet { @@ -162,7 +175,7 @@ fn make_job( .get_attr("system") .and_then(|v| v.as_string()) .unwrap_or_default(); - let outputs = output_paths(value); + let outputs = output_paths(store, &drv_path); let meta = if options.meta { read_meta(value) } else { None }; let constituents = read_constituents(value); @@ -338,9 +351,7 @@ fn read_input_drvs( // keys them by store-relative basename. Re-add the store prefix so keys are // absolute `.drv` paths, and expose the value as the output-name list to // match the `nix-eval-jobs` `inputDrvs` contract (`{drv: ["out", ...]}`). - let store_dir = store - .store_dir() - .unwrap_or_else(|_| "/nix/store".to_string()); + let store_dir = store_dir(store); // A derivation with no input derivations (e.g. a fixed-output fetch) // legitimately has no `inputs.drvs`, so an absent key is normal and not // logged. @@ -352,11 +363,7 @@ fn read_input_drvs( return map; }; for (key, value) in drvs { - let full_path = if key.starts_with('/') { - key.clone() - } else { - format!("{store_dir}/{key}") - }; + let full_path = absolute_store_path(&store_dir, key); let Some(outputs) = input_drv_outputs(value) else { warn!(drv_path = %full_path, "failed to parse inputDrvs outputs"); continue; @@ -371,52 +378,72 @@ fn input_drv_outputs(value: &serde_json::Value) -> Option> { serde_json::from_value(outputs.clone()).ok() } -/// Collect each output's store path from a derivation value. +/// Read each output's store path from the derivation's `.drv`. +/// +/// `outPath` carries derivation context, and the only string accessor the C API +/// exposes realises it, which tries to build the derivation and fails. /// /// # Returns /// -/// A map from output name to its resolved store path, or `None` when resolution -/// fails for an individual output. -fn output_paths(value: &Value<'_>) -> BTreeMap> { +/// A map from output name to store path, [`None`] for a floating +/// content-addressed output whose path is not known until it is built. +fn output_paths( + store: &Store, + drv_path: &StorePath, +) -> BTreeMap> { let mut map = BTreeMap::new(); - let Ok(list) = value.get_attr("outputs") else { - return map; + let drv = match store.read_derivation(drv_path) { + Ok(drv) => drv, + Err(e) => { + warn!(error = %e, "failed to read derivation for outputs"); + return map; + }, + }; + let json = match drv.to_json() { + Ok(json) => json, + Err(e) => { + warn!(error = %e, "failed to serialize derivation for outputs"); + return map; + }, + }; + let parsed = match serde_json::from_str::(&json) { + Ok(parsed) => parsed, + Err(e) => { + warn!(error = %e, "failed to parse derivation JSON for outputs"); + return map; + }, }; - let Ok(len) = list.list_len() else { + let Some(outputs) = + parsed.get("outputs").and_then(serde_json::Value::as_object) + else { + warn!("derivation JSON is missing its outputs"); return map; }; - for i in 0..len { - let Ok(name_val) = list.list_get(i) else { - continue; - }; - let Ok(name) = name_val.as_string() else { - continue; - }; - let path = output_path_for(value, &name); - map.insert(name, path); + + let store_dir = store_dir(store); + for (name, output) in outputs { + let path = output + .get("path") + .and_then(serde_json::Value::as_str) + .map(|path| absolute_store_path(&store_dir, path)); + map.insert(name.clone(), path); } map } -/// Resolve the store path of a single named output. -/// -/// Each output is exposed on the derivation as an attribute whose `outPath` is -/// the store path; for non-standard derivations the attribute is coerced -/// directly as a string or path. -/// -/// # Returns -/// -/// The output's store path, or `None` if the output attribute is missing or -/// cannot be coerced to a path. -fn output_path_for(value: &Value<'_>, name: &str) -> Option { - let out = value.get_attr(name).ok()?; - if let Ok(path) = out.get_attr("outPath").and_then(|v| v.as_string()) { - return Some(path); - } - if let Ok(s) = out.as_string() { - return Some(s); +fn store_dir(store: &Store) -> String { + store + .store_dir() + .unwrap_or_else(|_| "/nix/store".to_string()) +} + +/// `nix_derivation_to_json` keys paths by store-relative basename. +fn absolute_store_path(store_dir: &str, path: &str) -> String { + if path.starts_with('/') { + path.to_owned() + } else { + format!("{store_dir}/{path}") } - out.as_path().ok().map(|p| p.to_string_lossy().into_owned()) } /// Create a direct Nix GC root symlink for `drv_path`. diff --git a/crates/evix/src/worker_process.rs b/crates/evix/src/worker_process.rs index 15ee9de..8aef13f 100644 --- a/crates/evix/src/worker_process.rs +++ b/crates/evix/src/worker_process.rs @@ -42,7 +42,7 @@ pub(crate) struct WorkerProcess { proc: Child, stdin: Compat, stdout: Compat>, - stderr_task: JoinHandle>, + stderr_task: Option>>, _nix_options_file: Option, } @@ -105,7 +105,7 @@ impl WorkerProcess { proc: child, stdin, stdout, - stderr_task, + stderr_task: Some(stderr_task), _nix_options_file: nix_options_file, }; worker.read_ready().await?; @@ -130,18 +130,30 @@ impl WorkerProcess { pub(crate) async fn stop(&mut self) { let _ = write_client(&mut self.stdin, &ClientMessage::Shutdown).await; let _ = self.proc.wait().await; - let _ = (&mut self.stderr_task).await; + self.take_stderr().await; } pub(crate) async fn abort(&mut self) { let _ = self.proc.start_kill(); let _ = self.proc.wait().await; - let _ = (&mut self.stderr_task).await; + self.take_stderr().await; } pub(crate) async fn wait_for_restart(&mut self) { let _ = self.proc.wait().await; - let _ = (&mut self.stderr_task).await; + self.take_stderr().await; + } + + /// Await the stderr capture task and return its bounded tail. + /// + /// [`exit_error`](Self::exit_error) drains a failed worker before the caller + /// decides whether to stop or abort it, and a [`JoinHandle`] panics when + /// polled after completion. + async fn take_stderr(&mut self) -> String { + let Some(task) = self.stderr_task.take() else { + return String::new(); + }; + task.await.ok().and_then(Result::ok).unwrap_or_default() } async fn read_ready(&mut self) -> Result<()> { @@ -193,11 +205,7 @@ impl WorkerProcess { source: anyhow::Error, ) -> anyhow::Error { let status = self.proc.wait().await.ok(); - let stderr = (&mut self.stderr_task) - .await - .ok() - .and_then(Result::ok) - .unwrap_or_default(); + let stderr = self.take_stderr().await; let stderr = stderr.trim(); let mut message = format!( "evix worker {} failed while reading {phase} for {attr}: {source}", @@ -411,6 +419,47 @@ mod tests { } } + fn stub_worker(label: &str) -> WorkerProcess { + let mut child = Command::new("cat") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .expect("spawn stub worker"); + let stdin = child.stdin.take().expect("stub stdin").compat_write(); + let stdout = + BufReader::new(child.stdout.take().expect("stub stdout")).compat(); + let stderr = child.stderr.take().expect("stub stderr"); + let stderr_label = label.to_owned(); + let stderr_task = + tokio::spawn(async move { capture_stderr(stderr_label, stderr).await }); + + WorkerProcess { + label: label.to_owned(), + proc: child, + stdin, + stdout, + stderr_task: Some(stderr_task), + _nix_options_file: None, + } + } + + #[test] + fn lifecycle_calls_are_idempotent_after_stderr_is_drained() { + tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .unwrap() + .block_on(async { + let mut worker = stub_worker("idempotent"); + + worker.abort().await; + worker.stop().await; + worker.wait_for_restart().await; + }); + } + #[test] fn captured_stderr_keeps_only_bounded_tail() { tokio::runtime::Builder::new_current_thread() diff --git a/nix/tests/eval.nix b/nix/tests/eval.nix index 3be15ed..873eaed 100644 --- a/nix/tests/eval.nix +++ b/nix/tests/eval.nix @@ -8,7 +8,7 @@ if system == "aarch64-linux" then "x86_64-linux" else "aarch64-linux"; - localFlake = '' + localFlake = /* nix */ '' { outputs = { self }: { hydraJobs.${system} = { @@ -23,7 +23,7 @@ }; } ''; - remoteExpr = '' + remoteExpr = /* nix */ '' { recurseForDerivations = true; remote = { @@ -37,7 +37,7 @@ }; } ''; - distributedExpr = '' + distributedExpr = /* nix */ '' { recurseForDerivations = true; groupA = { @@ -75,7 +75,7 @@ }; } ''; - routedRemoteExpr = '' + routedRemoteExpr = /* nix */ '' { recurseForDerivations = true; native = { @@ -131,7 +131,7 @@ in }; }; - testScript = '' + testScript = /* python */ '' import os import shlex @@ -141,7 +141,7 @@ in REMOTE_EXPR = ${builtins.toJSON remoteExpr} DISTRIBUTED_EXPR = ${builtins.toJSON distributedExpr} ROUTED_REMOTE_EXPR = ${builtins.toJSON routedRemoteExpr} - CLIENT_EXPR = ${builtins.toJSON '' + CLIENT_EXPR = ${builtins.toJSON /* nix */ '' { label }: { recurseForDerivations = true; client = (derivation {