circus-evaluator: migrate evix 0.3.3 -> 1.0.2

Replace2 blocking `evaluate_cancellable` + `spawn_blocking` with the v1
async `Session::open(config).await?` + `session.stream()` drain loop.

Signed-off-by: NotAShelf <raf@notashelf.dev>
Change-Id: I6d77fd62f962d4b2ad9c7f42d8de44356a6a6964

Author: NotAShelf

Cargo.lock

@@ -140,7 +140,7 @@ harmonia-store-path-info       = { git = "https://github.com/nix-community/harmo
 harmonia-utils-hash            = { git = "https://github.com/nix-community/harmonia.git", rev = "12c6742560dd1ba1e66f5cc2f04cabd4e99ae754" }
 
 # evix, in-process Nix evaluator (nix-eval-jobs replacement) driven as a library.
-evix = { version = "0.3.3", features = [ "flake" ] }
+evix = { version = "1.0.2", features = [ "flake" ] }
 
 # Workspace Clippy lints. All workspace crates must inherit them in individual
 # manifests for subcrates. The below list is a rather pedantic list of lints

Cargo.toml

@@ -320,6 +320,7 @@ async fn evaluate_flake(
     show_input_drvs: true,
     override_inputs,
     nix_options: policy.nix_options(),
+    ..evix::Config::default()
   };
 
   eval_command::run_eval(evix_config, timeout, "flake").await
@@ -504,6 +505,7 @@ async fn evaluate_legacy(
     show_input_drvs: true,
     override_inputs: Vec::new(),
     nix_options: NixEvalPolicy::from(config).nix_options(),
+    ..evix::Config::default()
   };
 
   eval_command::run_eval(evix_config, timeout, "legacy").await

crates/evaluator/src/nix.rs

@@ -1,17 +1,11 @@
-use std::{
-  sync::{
-    Arc,
-    Mutex,
-    atomic::{AtomicBool, Ordering},
-  },
-  time::Duration,
-};
+use std::time::Duration;
 
 use circus_common::{CiError, error::Result};
 use circus_config::EvaluatorConfig;
+use futures::StreamExt as _;
 use tokio::process::Command;
 
-use super::{EvalResult, NixJob, nix_job_from_derivation};
+use super::{EvalResult, nix_job_from_derivation};
 
 /// Nix evaluation settings derived from the evaluator configuration, forwarded
 /// to evix as `(key, value)` options.
@@ -100,52 +94,33 @@ pub(super) async fn run_eval(
     "Starting evix evaluation with Nix options"
   );
 
-  let collected: Arc<Mutex<(Vec<NixJob>, usize)>> =
-    Arc::new(Mutex::new((Vec::new(), 0)));
-  let cancel = Arc::new(AtomicBool::new(false));
+  let session = evix::Session::open(config)
+    .await
+    .map_err(|e| evix_eval_failure(description, &format!("{e:#}")))?;
 
-  let sink_state = Arc::clone(&collected);
-  let cancel_eval = Arc::clone(&cancel);
-  let handle = tokio::task::spawn_blocking(move || {
-    evix::evaluate_cancellable(&config, &cancel_eval, move |event| {
-      match event {
+  let collect = async move {
+    let mut events = session.stream();
+    let mut jobs = Vec::new();
+    let mut error_count = 0usize;
+
+    while let Some(event) = events.next().await {
+      match event.map_err(|e| evix_eval_failure(description, &format!("{e:#}")))? {
         evix::Event::Derivation(drv) => {
-          sink_state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .0
-            .push(nix_job_from_derivation(drv));
+          jobs.push(nix_job_from_derivation(&drv));
         },
         evix::Event::Error(err) => {
           tracing::warn!(job = %err.attr, "evix reported error: {}", err.error);
-          sink_state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .1 += 1;
+          error_count += 1;
         },
         evix::Event::AttrSet { .. } => {},
       }
-      Ok(())
-    })
-  });
-
-  match tokio::time::timeout(timeout, handle).await {
-    Ok(joined) => {
-      joined
-        .map_err(|e| {
-          CiError::NixEval(format!("evix evaluation task failed to join: {e}"))
-        })?
-        .map_err(|e| evix_eval_failure(description, &format!("{e:#}")))?;
+    }
+    Ok::<EvalResult, CiError>(EvalResult { jobs, error_count })
+  };
 
-      let result = {
-        let guard = collected
-          .lock()
-          .unwrap_or_else(std::sync::PoisonError::into_inner);
-        EvalResult {
-          jobs:        guard.0.clone(),
-          error_count: guard.1,
-        }
-      };
+  match tokio::time::timeout(timeout, collect).await {
+    Ok(result) => {
+      let result = result?;
       if result.error_count > 0 {
         tracing::warn!(
           error_count = result.error_count,
@@ -157,13 +132,9 @@ pub(super) async fn run_eval(
       }
       Ok(result)
     },
-    Err(_elapsed) => {
-      // Ask the in-flight evaluation to stop; its workers exit cooperatively.
-      cancel.store(true, Ordering::Relaxed);
-      Err(CiError::Timeout(format!(
-        "Nix evaluation timed out after {timeout:?}"
-      )))
-    },
+    Err(_elapsed) => Err(CiError::Timeout(format!(
+      "Nix evaluation timed out after {timeout:?}"
+    ))),
   }
 }
 
@@ -352,6 +323,7 @@ mod policy_tests {
       show_input_drvs: false,
       override_inputs: Vec::new(),
       nix_options:     NixEvalPolicy::from(&config).nix_options(),
+      ..evix::Config::default()
     };
 
     assert_eq!(evix_config.nix_options, vec![