git.delta.rocks / fleet / refs/commits / 210037fd620b

difftreelog

refactor nix store api cleanup

ssxzyxlqYaroslav Bolyukin2026-06-15parent: #7bc4be6.patch.diff

7 files changed

modifiedcmds/fleet/src/main.rsdiffbeforeafterboth
--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -24,8 +24,7 @@
 #[cfg(feature = "indicatif")]
 use indicatif::{ProgressState, ProgressStyle};
 use nix_eval::{
-	add_file_to_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries,
-	init_tokio_for_nix,
+	eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
 };
 use opentelemetry::trace::TracerProvider;
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
@@ -33,6 +32,7 @@
 	OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,
 };
 use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};
+use tokio::task::spawn_blocking;
 use tracing::{Instrument, error, info, info_span};
 #[cfg(feature = "indicatif")]
 use tracing_indicatif::IndicatifLayer;
@@ -59,7 +59,8 @@
 				Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;
 			let span = info_span!("prefetching", name = %name);
 			tasks.push(async move {
-				let added = tokio::task::spawn_blocking(move || add_file_to_store(&name, &path))
+				let store = eval_store();
+				let added = spawn_blocking(move || store.add_file(&name, &path))
 					.instrument(span.clone())
 					.await??;
 				let _g = span.enter();
@@ -121,9 +122,7 @@
 		Opts::Prefetch(p) => p.run(config).await?,
 		Opts::Tf(t) => t.run(config).await?,
 		// TODO: actually parse commands before starting the async runtime
-		Opts::Complete(c) => {
-			tokio::task::spawn_blocking(move || c.run(RootOpts::command())).await?
-		}
+		Opts::Complete(c) => spawn_blocking(move || c.run(RootOpts::command())).await?,
 	};
 	Ok(())
 }
modifiedcrates/fleet-base/src/host.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/host.rs
+++ b/crates/fleet-base/src/host.rs
@@ -13,7 +13,7 @@
 use camino::{Utf8Path, Utf8PathBuf};
 use chrono::{DateTime, Utc};
 use fleet_shared::SecretData;
-use nix_eval::{Store, Value, nix_go, nix_go_json, util::assert_warn};
+use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};
 use remowt_client::{AgentBundle, Remowt};
 use remowt_endpoints::fs::FsClient;
 use remowt_link_shared::Address;
@@ -427,8 +427,10 @@
 		let store = self.nix_store().await?;
 		{
 			let path = path.clone();
-			spawn_blocking(move || nix_eval::copy_closure_to(&store, path.as_ref()))
-				.await?
+			let store = eval_store();
+			spawn_blocking(move || store.copy_to(&store, path.as_ref()))
+				.await
+				.expect("copy_to panicked")
 				.context("copying closure to remote store")?;
 		}
 		Ok(path)
modifiedcrates/nix-eval/src/drv.rsdiffbeforeafterboth
--- a/crates/nix-eval/src/drv.rs
+++ b/crates/nix-eval/src/drv.rs
@@ -1,41 +1,21 @@
 use std::collections::{HashMap, HashSet, VecDeque};
-use std::ffi::CString;
 
 use anyhow::{Result, bail};
+use camino::{Utf8Component, Utf8Path, Utf8PathBuf};
 use serde::Deserialize;
 
 use crate::nix_raw::{derivation_free, derivation_to_json, store_drv_from_store_path};
-use crate::{copy_nix_str, with_store_context};
-
-fn store_dir() -> Result<String> {
-	let mut out = String::new();
-	with_store_context(|c, store, _| unsafe {
-		crate::nix_raw::store_get_storedir(c, store, Some(copy_nix_str), (&raw mut out).cast())
-	})?;
-	Ok(out)
-}
-
-fn to_absolute_store_path(store_dir: &str, path: &str) -> String {
-	if path.starts_with('/') {
-		path.to_owned()
-	} else {
-		format!("{store_dir}/{path}")
-	}
-}
+use crate::{Store, copy_nix_str, with_default_context};
 
 pub struct Derivation(*mut crate::nix_raw::derivation);
 unsafe impl Send for Derivation {}
 
 impl Derivation {
-	pub fn from_path(drv_path: &str) -> Result<Self> {
-		let path_c = CString::new(drv_path)?;
-		let store_path = with_store_context(|c, store, _| unsafe {
-			crate::nix_raw::store_parse_path(c, store, path_c.as_ptr())
-		})?;
-		let drv = with_store_context(|c, store, _| unsafe {
-			store_drv_from_store_path(c, store, store_path)
+	pub fn from_path(store: &Store, drv_path: &Utf8Path) -> Result<Self> {
+		let store_path = store.parse_path(drv_path)?;
+		let drv = with_default_context(|c, _| unsafe {
+			store_drv_from_store_path(c, store.as_ptr(), store_path.as_ptr())
 		});
-		unsafe { crate::nix_raw::store_path_free(store_path) };
 		let drv = drv?;
 		if drv.is_null() {
 			bail!("failed to read derivation from {drv_path}");
@@ -45,7 +25,7 @@
 
 	pub fn to_json_string(&self) -> Result<String> {
 		let mut out = String::new();
-		with_store_context(|c, _, _| unsafe {
+		with_default_context(|c, _| unsafe {
 			derivation_to_json(c, self.0, Some(copy_nix_str), (&raw mut out).cast())
 		})?;
 		Ok(out)
@@ -78,9 +58,9 @@
 #[derive(Debug, Deserialize)]
 pub struct DrvInputs {
 	#[serde(default)]
-	pub srcs: Vec<String>,
+	pub srcs: Vec<Utf8PathBuf>,
 	#[serde(default)]
-	pub drvs: HashMap<String, DrvInputEntry>,
+	pub drvs: HashMap<Utf8PathBuf, DrvInputEntry>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -90,23 +70,23 @@
 
 #[derive(Debug, Clone)]
 pub struct DrvGraph {
-	pub root: String,
-	pub nodes: HashMap<String, DrvNode>,
+	pub root: Utf8PathBuf,
+	pub nodes: HashMap<Utf8PathBuf, DrvNode>,
 }
 
 #[derive(Debug, Clone)]
 pub struct DrvNode {
 	pub name: String,
-	pub input_drvs: HashMap<String, Vec<String>>,
-	pub input_srcs: Vec<String>,
+	pub input_drvs: HashMap<Utf8PathBuf, Vec<String>>,
+	pub input_srcs: Vec<Utf8PathBuf>,
 	// TODO: CA outputs without a known paths are skipped
-	pub outputs: HashMap<String, String>,
+	pub outputs: HashMap<String, Utf8PathBuf>,
 }
 
 impl DrvGraph {
-	pub fn resolve(drv_path: &str) -> Result<Self> {
-		let sd = store_dir()?;
-		let root = to_absolute_store_path(&sd, drv_path);
+	pub fn resolve(store: &Store, drv_path: &Utf8Path) -> Result<Self> {
+		let sd = store.store_dir()?;
+		let root = sd.join(drv_path);
 
 		let mut nodes = HashMap::new();
 		let mut queue = VecDeque::new();
@@ -115,14 +95,14 @@
 		visited.insert(root.clone());
 
 		while let Some(path) = queue.pop_front() {
-			let drv = Derivation::from_path(&path)?;
+			let drv = Derivation::from_path(store, &path)?;
 			let parsed = drv.parsed()?;
 
-			let input_drvs: HashMap<String, Vec<String>> = parsed
+			let input_drvs: HashMap<Utf8PathBuf, Vec<String>> = parsed
 				.inputs
 				.drvs
 				.into_iter()
-				.map(|(k, v)| (to_absolute_store_path(&sd, &k), v.outputs))
+				.map(|(k, v)| (sd.join(&k), v.outputs))
 				.collect();
 
 			for dep_path in input_drvs.keys() {
@@ -131,10 +111,10 @@
 				}
 			}
 
-			let outputs: HashMap<String, String> = parsed
+			let outputs: HashMap<String, Utf8PathBuf> = parsed
 				.outputs
 				.into_iter()
-				.filter_map(|(name, out)| out.path.map(|p| (name, to_absolute_store_path(&sd, &p))))
+				.filter_map(|(name, out)| out.path.map(|p| (name, sd.join(&p))))
 				.collect();
 
 			nodes.insert(
@@ -151,11 +131,11 @@
 		Ok(Self { root, nodes })
 	}
 
-	pub fn wanted_outputs(&self, root_outputs: &[String]) -> HashMap<String, Vec<String>> {
-		let mut wanted: HashMap<String, HashSet<String>> = HashMap::new();
+	pub fn wanted_outputs(&self, root_outputs: &[String]) -> HashMap<Utf8PathBuf, Vec<String>> {
+		let mut wanted: HashMap<Utf8PathBuf, HashSet<String>> = HashMap::new();
 		wanted.insert(self.root.clone(), root_outputs.iter().cloned().collect());
 
-		let mut queue: VecDeque<String> = VecDeque::new();
+		let mut queue: VecDeque<Utf8PathBuf> = VecDeque::new();
 		queue.push_back(self.root.clone());
 		while let Some(path) = queue.pop_front() {
 			let Some(node) = self.nodes.get(&path) else {
@@ -186,12 +166,19 @@
 	}
 }
 
-fn extract_drv_name(drv_path: &str) -> String {
-	drv_path
-		.rsplit('/')
+pub fn extract_drv_name(drv_path: &Utf8Path) -> String {
+	let comp = drv_path
+		.components()
+		.rev()
 		.next()
-		.and_then(|f| f.strip_suffix(".drv"))
-		.and_then(|f| f.split_once('-').map(|(_, name)| name))
-		.unwrap_or(drv_path)
-		.to_owned()
+		.expect("drv path is at least one component");
+	let Utf8Component::Normal(n) = comp else {
+		panic!("drv path is normal");
+	};
+
+	let n = n.strip_suffix(".drv").unwrap_or(n);
+
+	let n = n.split_once(' ').map(|(_, n)| n).unwrap_or(n);
+
+	n.to_owned()
 }
modifiedcrates/nix-eval/src/lib.rsdiffbeforeafterboth
--- a/crates/nix-eval/src/lib.rs
+++ b/crates/nix-eval/src/lib.rs
@@ -25,7 +25,7 @@
 	PrimOpFun, Store as c_store, StorePath as c_store_path, alloc_primop, alloc_value,
 	bindings_builder_free, bindings_builder_insert, c_context, c_context_create, c_context_free,
 	clear_err, copy_value, err_NIX_ERR_KEY, err_NIX_ERR_NIX_ERROR, err_NIX_ERR_OVERFLOW,
-	err_NIX_ERR_UNKNOWN, err_code, err_info_msg, err_msg, eval_state_build,
+	err_NIX_ERR_UNKNOWN, err_NIX_OK, err_code, err_info_msg, err_msg, eval_state_build,
 	eval_state_builder_load, eval_state_builder_new, eval_state_builder_set_eval_setting,
 	expr_eval_from_string, fetchers_settings, fetchers_settings_free, fetchers_settings_new,
 	flake_lock, flake_lock_flags, flake_lock_flags_free, flake_lock_flags_new, flake_reference,
@@ -320,15 +320,16 @@
 struct GlobalState {
 	// Store should be valid as long as EvalState is valid
 	#[allow(dead_code)]
-	store: Store,
+	store: Arc<Store>,
 	state: EvalState,
 }
 impl GlobalState {
 	fn new() -> Result<Self> {
 		let mut ctx = NixContext::new();
-		let store = ctx
-			.run_in_context(|c| unsafe { store_open(c, c"auto".as_ptr(), null_mut()) })
-			.map(Store)?;
+		let store = Arc::new(
+			ctx.run_in_context(|c| unsafe { store_open(c, c"auto".as_ptr(), null_mut()) })
+				.map(Store)?,
+		);
 
 		let builder = ctx.run_in_context(|c| unsafe { eval_state_builder_new(c, store.0) })?;
 		ctx.run_in_context(|c| unsafe { eval_state_builder_load(c, builder) })?;
@@ -385,66 +386,8 @@
 	v
 }
 
-/// Same as with_default_context, but also passes store...
-/// Yep, this code is garbage and needs to be refactored.
-pub(crate) fn with_store_context<T>(
-	f: impl FnOnce(*mut c_context, *mut c_store, *mut c_eval_state) -> T,
-) -> Result<T> {
-	let global = &GLOBAL_STATE;
-	let (ctx, store, state) =
-		THREAD_STATE.with_borrow_mut(|w| (w.ctx.0, global.store.0, global.state.0));
-	let mut ctx = NixContext(ctx);
-	let v = ctx.run_in_context(|c| f(c, store, state));
-	std::mem::forget(ctx);
-	v
-}
-
 pub fn set_setting(s: &CStr, v: &CStr) -> Result<()> {
 	with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())
-}
-
-#[instrument(skip(dst))]
-pub fn copy_closure_to(dst: &Store, path: &Utf8Path) -> Result<()> {
-	let path_c = CString::new(path.as_str())?;
-	with_store_context(|c, src_store, _state| -> Result<()> {
-		let sp = unsafe { store_parse_path(c, src_store, path_c.as_ptr()) };
-		if sp.is_null() {
-			bail!("failed to parse store path {path}");
-		}
-		let rc = unsafe { store_copy_closure(c, src_store, dst.0, sp) };
-		unsafe { store_path_free(sp) };
-		if rc != nix_raw::err_NIX_OK {
-			bail!("store_copy_closure failed (code {rc})");
-		}
-		Ok(())
-	})?
-}
-
-#[instrument]
-pub fn switch_profile(profile: &str, store_path: &Utf8Path) -> Result<()> {
-	let msg = with_store_context(|_c, store, _state| unsafe {
-		nix_cxx::switch_profile(store.cast(), profile, store_path.as_str())
-	})?
-	.to_string();
-	if msg.is_empty() {
-		Ok(())
-	} else {
-		bail!("failed to switch profile {profile}: {msg}");
-	}
-}
-
-// TODO: fleet operator-managed key file
-#[instrument]
-pub fn sign_closure(store_path: &str, key_file: &str) -> Result<()> {
-	let msg = with_store_context(|_c, store, _state| unsafe {
-		nix_cxx::sign_closure(store.cast(), store_path, key_file)
-	})?
-	.to_string();
-	if msg.is_empty() {
-		Ok(())
-	} else {
-		bail!("failed to sign {store_path}: {msg}");
-	}
 }
 
 #[derive(Debug)]
@@ -480,48 +423,8 @@
 			current: g.current,
 		})
 		.collect())
-}
-
-#[instrument]
-pub fn add_file_to_store(name: &str, path: &Utf8Path) -> Result<AddedFile> {
-	let res = with_store_context(|_c, store, _state| unsafe {
-		nix_cxx::add_file_to_store(store.cast(), name, path.as_str())
-	})?;
-	if !res.error.is_empty() {
-		bail!("failed to add {path} to store: {}", res.error);
-	}
-	Ok(AddedFile {
-		store_path: Utf8PathBuf::from(res.store_path),
-		hash: res.hash,
-	})
-}
-
-pub fn build_drv_outputs(drv_path: &str, output_names: &[String]) -> Result<Vec<String>> {
-	let joined = output_names.join("\n");
-	let res = with_store_context(|_c, store, _state| unsafe {
-		nix_cxx::build_drv_outputs(store.cast(), drv_path, &joined)
-	})?;
-	if !res.error.is_empty() {
-		bail!("build of {drv_path} failed: {}", res.error);
-	}
-	Ok(res.outputs)
 }
 
-pub fn substitute_paths(paths: &[String]) -> Result<Vec<String>> {
-	let joined = paths.join("\n");
-	let res = with_store_context(|_c, store, _state| unsafe {
-		nix_cxx::substitute_paths(store.cast(), &joined)
-	})?;
-	if !res.error.is_empty() {
-		warn!("substitute_paths reported: {}", res.error);
-	}
-	Ok(res.outputs)
-}
-
-pub fn is_valid_path(path: &str) -> Result<bool> {
-	with_store_context(|_c, store, _state| unsafe { nix_cxx::is_valid_path(store.cast(), path) })
-}
-
 pub struct FetchSettings(*mut fetchers_settings);
 impl FetchSettings {
 	pub fn new() -> Self {
@@ -624,6 +527,10 @@
 unsafe impl Send for Store {}
 unsafe impl Sync for Store {}
 
+pub fn eval_store() -> Arc<Store> {
+	GLOBAL_STATE.store.clone()
+}
+
 impl Store {
 	pub fn open(uri: &str) -> Result<Self> {
 		let uri = CString::new(uri)?;
@@ -634,11 +541,108 @@
 		Ok(Store(ptr))
 	}
 
-	fn parse_path(&self, path: &CStr) -> Result<StorePath> {
+	pub fn parse_path(&self, path: &Utf8Path) -> Result<StorePath> {
+		let path = CString::new(path.as_str()).expect("valid cstr");
 		with_default_context(|c, _| {
 			StorePath(unsafe { store_parse_path(c, self.0, path.as_ptr()) })
 		})
 	}
+
+	#[instrument(skip(self))]
+	pub fn sign_closure(&self, path: &Utf8Path, key_file: &Utf8Path) -> Result<()> {
+		let err = with_default_context(|_, _| unsafe {
+			nix_cxx::sign_closure(self.as_ptr().cast(), path.as_str(), key_file.as_str())
+		})?
+		.to_string();
+
+		if err.is_empty() {
+			Ok(())
+		} else {
+			bail!("failed to sign {path}: {err}");
+		}
+	}
+
+	#[instrument(skip(self, dst))]
+	pub fn copy_to(&self, dst: &Store, path: &Utf8Path) -> Result<()> {
+		let sp = self
+			.parse_path(&path)
+			.context("failed to parse store path")?;
+		let rc = with_default_context(|c, _| unsafe {
+			store_copy_closure(c, self.as_ptr(), dst.0, sp.as_ptr())
+		})?;
+		if rc != err_NIX_OK {
+			bail!("store_copy_closure failed (code {rc})");
+		}
+		Ok(())
+	}
+
+	/// Would only work with local store.
+	#[instrument(skip(self))]
+	pub fn switch_profile(&self, profile: &str, path: &Utf8Path) -> Result<()> {
+		let msg = unsafe { nix_cxx::switch_profile(self.as_ptr().cast(), profile, path.as_str()) };
+		if msg.is_empty() {
+			Ok(())
+		} else {
+			bail!("failed to switch profile {profile}: {msg}");
+		}
+	}
+
+	#[instrument(skip(self))]
+	pub fn add_file(&self, name: &str, path: &Utf8Path) -> Result<AddedFile> {
+		let msg = unsafe { nix_cxx::add_file_to_store(self.as_ptr().cast(), name, path.as_str()) };
+		if !msg.error.is_empty() {
+			bail!("failed to add {path} to store: {}", msg.error)
+		}
+		Ok(AddedFile {
+			store_path: Utf8PathBuf::from(msg.store_path),
+			hash: msg.hash,
+		})
+	}
+
+	#[instrument(skip(self))]
+	pub fn substitute_paths(&self, paths: &[Utf8PathBuf]) -> Result<Vec<Utf8PathBuf>> {
+		let joined = paths.into_iter().join("\n");
+		let res = unsafe { nix_cxx::substitute_paths(self.as_ptr().cast(), &joined) };
+		if !res.error.is_empty() {
+			warn!("substitute_paths reported: {}", res.error);
+		}
+		Ok(res.outputs.into_iter().map(Utf8PathBuf::from).collect())
+	}
+
+	#[instrument(skip(self))]
+	pub fn is_valid_path(&self, path: &Utf8Path) -> bool {
+		unsafe { nix_cxx::is_valid_path(self.as_ptr().cast(), path.as_str()) }
+	}
+
+	#[instrument(skip(self))]
+	pub fn build_drv_outputs(
+		&self,
+		drv_path: &Utf8Path,
+		output_names: &[String],
+	) -> Result<Vec<String>> {
+		let joined = output_names.join("\n");
+		let res =
+			unsafe { nix_cxx::build_drv_outputs(self.as_ptr().cast(), drv_path.as_str(), &joined) };
+		if !res.error.is_empty() {
+			bail!("build of {drv_path} failed: {}", res.error);
+		}
+		Ok(res.outputs)
+	}
+
+	#[instrument(skip(self))]
+	pub fn store_dir(&self) -> Result<Utf8PathBuf> {
+		let mut out = String::new();
+		with_default_context(|c, es| unsafe {
+			nix_raw::store_get_storedir(c, self.as_ptr(), Some(copy_nix_str), (&raw mut out).cast())
+		})?;
+		let p = Utf8PathBuf::from(out);
+		assert!(p.is_absolute());
+		Ok(p)
+	}
+
+	fn as_ptr(&self) -> *mut c_store {
+		self.0
+	}
 }
 impl Drop for Store {
 	fn drop(&mut self) {
@@ -1060,11 +1064,12 @@
 			self.clone()
 		};
 
-		let drv_path = v
-			.get_field("drvPath")
-			.context("getting drvPath")?
-			.to_string()?;
-		let graph = Arc::new(drv::DrvGraph::resolve(&drv_path)?);
+		let drv_path = Utf8PathBuf::from(
+			v.get_field("drvPath")
+				.context("getting drvPath")?
+				.to_string()?,
+		);
+		let graph = Arc::new(drv::DrvGraph::resolve(&eval_store(), &drv_path)?);
 		let _guard = logging::register_build_graph(&Span::current(), &graph);
 
 		scheduler::build_graph_sync(graph.clone(), vec![output.to_owned()])?;
@@ -1255,8 +1260,12 @@
 	}
 }
 
-struct StorePath(*mut c_store_path);
-impl StorePath {}
+pub struct StorePath(*mut c_store_path);
+impl StorePath {
+	fn as_ptr(&self) -> *mut c_store_path {
+		self.0
+	}
+}
 
 impl Drop for StorePath {
 	fn drop(&mut self) {
modifiedcrates/nix-eval/src/logging.rsdiffbeforeafterboth
--- a/crates/nix-eval/src/logging.rs
+++ b/crates/nix-eval/src/logging.rs
@@ -2,6 +2,7 @@
 use std::fmt::Arguments;
 use std::sync::{LazyLock, Mutex};
 
+use camino::{Utf8Path, Utf8PathBuf};
 use cxx::ExternType;
 use tracing::{
 	Level, Span, debug, debug_span, error, error_span, info, info_span, trace, trace_span, warn,
@@ -11,6 +12,8 @@
 use tracing_indicatif::span_ext::IndicatifSpanExt as _;
 use vte::Parser;
 
+use crate::drv::extract_drv_name;
+
 #[derive(Debug)]
 enum ActivityType {
 	Unknown = 0,
@@ -33,20 +36,13 @@
 	a.strip_prefix(pref)?.strip_suffix(suff)
 }
 
-fn parse_path(path: &str) -> &str {
-	strip_prefix_suffix(path, "\x1b[35;1m", "\x1b[0m").unwrap_or(path)
+fn parse_path(path: &str) -> Utf8PathBuf {
+	Utf8PathBuf::from(strip_prefix_suffix(path, "\x1b[35;1m", "\x1b[0m").unwrap_or(path))
 }
 
-fn parse_drv(drv: &str) -> &str {
+fn parse_drv(drv: &str) -> String {
 	let drv = parse_path(drv);
-	if let Some(pkg) = drv.strip_prefix("/nix/store/") {
-		let mut it = pkg.splitn(2, '-');
-		it.next();
-		if let Some(pkg) = it.next() {
-			return pkg;
-		}
-	}
-	drv
+	extract_drv_name(&drv)
 }
 fn parse_host(host: &str) -> &str {
 	if host.is_empty() || host == "local" {
@@ -287,19 +283,19 @@
 
 struct DrvGraphEntry {
 	name: String,
-	parent: Option<String>,
+	parent: Option<Utf8PathBuf>,
 	span: Option<Span>,
 	refcount: usize,
 }
 
-static DRV_GRAPH: LazyLock<Mutex<HashMap<String, DrvGraphEntry>>> =
+static DRV_GRAPH: LazyLock<Mutex<HashMap<Utf8PathBuf, DrvGraphEntry>>> =
 	LazyLock::new(|| Mutex::new(HashMap::new()));
 
-static ACTIVITY_TO_DRV: LazyLock<Mutex<HashMap<u64, String>>> =
+static ACTIVITY_TO_DRV: LazyLock<Mutex<HashMap<u64, Utf8PathBuf>>> =
 	LazyLock::new(|| Mutex::new(HashMap::new()));
 
 pub struct BuildGraphGuard {
-	paths: Vec<String>,
+	paths: Vec<Utf8PathBuf>,
 }
 
 impl Drop for BuildGraphGuard {
@@ -369,7 +365,7 @@
 	BuildGraphGuard { paths }
 }
 
-fn ensure_drv_span(drv_path: &str) -> Option<Span> {
+fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {
 	let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");
 
 	if let Some(span) = drv_graph.get(drv_path).and_then(|e| e.span.clone()) {
@@ -442,7 +438,7 @@
 			self.fields.first().and_then(|f| match f {
 				FieldValue::Str(drv_path) => {
 					let clean = parse_path(drv_path);
-					let span = ensure_drv_span(clean);
+					let span = ensure_drv_span(&clean);
 					if span.is_some() {
 						ACTIVITY_TO_DRV
 							.lock()
modifiedcrates/nix-eval/src/scheduler.rsdiffbeforeafterboth
before · crates/nix-eval/src/scheduler.rs
1use std::collections::{HashMap, HashSet};2use std::sync::Arc;34use anyhow::{Context, Result, bail};5use futures::stream::{FuturesUnordered, StreamExt};6use tokio::sync::{Semaphore, broadcast};7use tracing::{debug, info, instrument, warn};89use crate::drv::DrvGraph;1011#[derive(Clone, Debug)]12pub enum BuildEvent {13	SubstitutePrepassStarted {14		paths: usize,15	},16	SubstitutePrepassFinished {17		satisfied: usize,18	},19	DrvStarted {20		drv_path: String,21		name: String,22		wanted: Vec<String>,23	},24	DrvSkipped {25		drv_path: String,26		name: String,27	},28	DrvFinished {29		drv_path: String,30		name: String,31	},32	DrvFailed {33		drv_path: String,34		name: String,35		error: String,36	},37	DrvCancelled {38		drv_path: String,39		name: String,40		failed_dep: String,41	},42}4344pub struct Scheduler {45	parallelism: usize,46	events: broadcast::Sender<BuildEvent>,47}4849impl Scheduler {50	pub fn new(parallelism: usize) -> Self {51		let parallelism = parallelism.max(1);52		let (events, _) = broadcast::channel(1024);53		Self {54			parallelism,55			events,56		}57	}5859	pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {60		self.events.subscribe()61	}6263	#[instrument(name = "scheduler", skip(self, graph), fields(root = %graph.root, nodes = graph.nodes.len()))]64	pub async fn run(&self, graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {65		let wanted = graph.wanted_outputs(&root_outputs);6667		self.substitute_prepass(&graph, &wanted).await?;68		self.build_topo(&graph, wanted).await69	}7071	async fn substitute_prepass(72		&self,73		graph: &DrvGraph,74		wanted: &HashMap<String, Vec<String>>,75	) -> Result<()> {76		let paths = collect_substitute_paths(graph, wanted);77		if paths.is_empty() {78			return Ok(());79		}80		let _ = self81			.events82			.send(BuildEvent::SubstitutePrepassStarted { paths: paths.len() });83		debug!("substitute pre-pass: {} paths", paths.len());8485		let satisfied = tokio::task::spawn_blocking(move || crate::substitute_paths(&paths))86			.await87			.expect("substitute pre-pass task should not panic")?;8889		let _ = self.events.send(BuildEvent::SubstitutePrepassFinished {90			satisfied: satisfied.len(),91		});92		Ok(())93	}9495	async fn build_topo(96		&self,97		graph: &Arc<DrvGraph>,98		wanted: HashMap<String, Vec<String>>,99	) -> Result<()> {100		let mut indeg: HashMap<String, usize> = graph101			.nodes102			.iter()103			.map(|(k, n)| (k.clone(), n.input_drvs.len()))104			.collect();105		let mut dependents: HashMap<String, Vec<String>> = HashMap::new();106		for (path, node) in &graph.nodes {107			for dep in node.input_drvs.keys() {108				dependents109					.entry(dep.clone())110					.or_default()111					.push(path.clone());112			}113		}114115		let sem = Arc::new(Semaphore::new(self.parallelism));116		let mut ready: Vec<String> = indeg117			.iter()118			.filter(|(_, d)| **d == 0)119			.map(|(k, _)| k.clone())120			.collect();121		let mut in_flight = FuturesUnordered::new();122		let mut failed: HashMap<String, String> = HashMap::new();123		// Tainted = transitively depends on a failed drv124		let mut tainted: HashMap<String, String> = HashMap::new();125126		loop {127			let batch: Vec<String> = std::mem::take(&mut ready);128			for path in batch {129				if let Some(failed_dep) = tainted.get(&path) {130					let name = graph131						.nodes132						.get(&path)133						.map(|n| n.name.clone())134						.unwrap_or_default();135					let _ = self.events.send(BuildEvent::DrvCancelled {136						drv_path: path.clone(),137						name,138						failed_dep: failed_dep.clone(),139					});140					propagate_done(&dependents, &mut indeg, &mut ready, &path);141					continue;142				}143144				let sem = sem.clone();145				let events = self.events.clone();146				let graph = graph.clone();147				let wanted_here = wanted.get(&path).cloned().unwrap_or_default();148				in_flight.push(tokio::spawn(async move {149					let _permit = sem.acquire_owned().await.expect("semaphore not closed");150					let node = graph151						.nodes152						.get(&path)153						.expect("ready node must be in graph")154						.clone();155					let name = node.name.clone();156157					let all_valid = !wanted_here.is_empty()158						&& wanted_here.iter().all(|o| {159							node.outputs160								.get(o)161								.map(|p| crate::is_valid_path(p).unwrap_or(false))162								.unwrap_or(false)163						});164					if all_valid {165						let _ = events.send(BuildEvent::DrvSkipped {166							drv_path: path.clone(),167							name: name.clone(),168						});169						return (path, name, Ok::<(), anyhow::Error>(()));170					}171172					let _ = events.send(BuildEvent::DrvStarted {173						drv_path: path.clone(),174						name: name.clone(),175						wanted: wanted_here.clone(),176					});177178					let path_for_build = path.clone();179					let res = tokio::task::spawn_blocking(move || {180						crate::build_drv_outputs(&path_for_build, &wanted_here)181					})182					.await183					.expect("build task should not panic");184185					match res {186						Ok(_) => {187							let _ = events.send(BuildEvent::DrvFinished {188								drv_path: path.clone(),189								name: name.clone(),190							});191							(path, name, Ok(()))192						}193						Err(e) => {194							let msg = format!("{e:#}");195							let _ = events.send(BuildEvent::DrvFailed {196								drv_path: path.clone(),197								name: name.clone(),198								error: msg,199							});200							(path, name, Err(e))201						}202					}203				}));204			}205206			let Some(joined) = in_flight.next().await else {207				break;208			};209			let (finished, _name, res) = match joined {210				Ok(t) => t,211				Err(e) => bail!("scheduler task panicked: {e}"),212			};213			match res {214				Ok(()) => {215					propagate_done(&dependents, &mut indeg, &mut ready, &finished);216				}217				Err(e) => {218					failed.insert(finished.clone(), format!("{e:#}"));219					mark_tainted(&dependents, &finished, &mut tainted);220					propagate_done(&dependents, &mut indeg, &mut ready, &finished);221				}222			}223		}224225		let stuck: Vec<_> = indeg226			.iter()227			.filter(|(_, d)| **d != 0)228			.map(|(k, _)| k.as_str())229			.collect();230		if !stuck.is_empty() {231			warn!(232				"scheduler finished with {} nodes still pending (loop?)",233				stuck.len()234			);235		}236237		if failed.is_empty() {238			info!("scheduler completed");239			Ok(())240		} else {241			let mut report = format!("{} drv(s) failed to build:", failed.len());242			let mut sorted: Vec<_> = failed.iter().collect();243			sorted.sort_by(|a, b| a.0.cmp(b.0));244			for (path, err) in sorted {245				let name = graph246					.nodes247					.get(path)248					.map(|n| n.name.as_str())249					.unwrap_or("?");250				let chain = path_to_root(graph, path);251				report.push_str(&format!(252					"\n\n  {name} ({path}):\n    {err}\n    needed by: {}",253					chain.join(" => "),254				));255			}256			Err(anyhow::anyhow!(report))257		}258	}259}260261fn propagate_done(262	dependents: &HashMap<String, Vec<String>>,263	indeg: &mut HashMap<String, usize>,264	ready: &mut Vec<String>,265	finished: &str,266) {267	if let Some(deps) = dependents.get(finished) {268		for d in deps {269			let entry = indeg.get_mut(d).expect("dependent must have indeg");270			*entry = entry.saturating_sub(1);271			if *entry == 0 {272				ready.push(d.clone());273			}274		}275	}276}277278fn mark_tainted(279	dependents: &HashMap<String, Vec<String>>,280	failed: &str,281	tainted: &mut HashMap<String, String>,282) {283	let mut queue: Vec<String> = dependents.get(failed).cloned().unwrap_or_default();284	while let Some(node) = queue.pop() {285		if tainted286			.entry(node.clone())287			.or_insert_with(|| failed.to_owned())288			== failed289		{290			if let Some(deps) = dependents.get(&node) {291				for d in deps {292					if !tainted.contains_key(d) {293						queue.push(d.clone());294					}295				}296			}297		}298	}299}300301fn path_to_root(graph: &DrvGraph, from: &str) -> Vec<String> {302	let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();303	for (path, node) in &graph.nodes {304		for dep in node.input_drvs.keys() {305			dependents306				.entry(dep.as_str())307				.or_default()308				.push(path.as_str());309		}310	}311312	let mut chain: Vec<String> = vec![node_name(graph, from)];313	let mut cur = from;314	let mut seen: HashSet<&str> = HashSet::new();315	seen.insert(cur);316	while cur != graph.root.as_str() {317		let Some(next) = dependents.get(cur).and_then(|v| v.first().copied()) else {318			break;319		};320		if !seen.insert(next) {321			break;322		}323		chain.push(node_name(graph, next));324		cur = next;325	}326	chain327}328329fn node_name(graph: &DrvGraph, path: &str) -> String {330	graph331		.nodes332		.get(path)333		.map(|n| n.name.clone())334		.unwrap_or_else(|| path.to_owned())335}336337fn collect_substitute_paths(338	graph: &DrvGraph,339	wanted: &HashMap<String, Vec<String>>,340) -> Vec<String> {341	let mut paths: HashSet<String> = HashSet::new();342	for node in graph.nodes.values() {343		for src in &node.input_srcs {344			paths.insert(src.clone());345		}346	}347	for (path, outs) in wanted {348		let Some(node) = graph.nodes.get(path) else {349			continue;350		};351		for o in outs {352			if let Some(p) = node.outputs.get(o) {353				paths.insert(p.clone());354			}355		}356	}357	let mut v: Vec<_> = paths.into_iter().collect();358	v.sort();359	v360}361362// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing363// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one364// will work as a hard cap.365pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {366	let parallelism = std::thread::available_parallelism()367		.map(|p| p.get())368		.unwrap_or(4);369	let scheduler = Scheduler::new(parallelism);370	crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })371		.context("scheduler run")372}
modifiedcrates/remowt-fleet/src/lib.rsdiffbeforeafterboth
--- a/crates/remowt-fleet/src/lib.rs
+++ b/crates/remowt-fleet/src/lib.rs
@@ -1,13 +1,15 @@
 use std::path::PathBuf;
 
 use anyhow::{Context as _, Result};
+use bifrostlink::declarative::endpoints;
 use bifrostlink::Config;
-use bifrostlink::declarative::endpoints;
 use camino::Utf8PathBuf;
+use nix_eval::eval_store;
 use remowt_client::Remowt;
 use remowt_endpoints::nix_daemon::NixDaemonClient;
 use serde::{Deserialize, Serialize};
 use tokio::net::UnixListener;
+use tokio::task::spawn_blocking;
 use tracing::error;
 
 pub struct Nix;
@@ -35,9 +37,10 @@
 		profile: String,
 		store_path: Utf8PathBuf,
 	) -> Result<(), NixError> {
-		tokio::task::spawn_blocking(move || nix_eval::switch_profile(&profile, &store_path))
+		let store = eval_store();
+		spawn_blocking(move || store.switch_profile(&profile, &store_path))
 			.await
-			.map_err(|e| NixError::Profile(e.to_string()))?
+			.expect("switch_profile panicked")
 			.map_err(|e| NixError::Profile(e.to_string()))
 	}
 
@@ -47,11 +50,12 @@
 		store_path: Utf8PathBuf,
 		key_file: Utf8PathBuf,
 	) -> Result<(), NixError> {
-		tokio::task::spawn_blocking(move || {
-			nix_eval::sign_closure(store_path.as_str(), key_file.as_str())
+		spawn_blocking(move || {
+			let store = eval_store();
+			store.sign_closure(&store_path, &key_file)
 		})
 		.await
-		.map_err(|e| NixError::Sign(e.to_string()))?
+		.expect("store signing panicked")
 		.map_err(|e| NixError::Sign(e.to_string()))
 	}
 
@@ -60,11 +64,11 @@
 		&self,
 		profile: String,
 	) -> Result<Vec<nix_eval::ProfileGeneration>, NixError> {
-		tokio::task::spawn_blocking(move || {
+		spawn_blocking(move || {
 			nix_eval::list_generations(&format!("/nix/var/nix/profiles/{profile}"))
 		})
 		.await
-		.map_err(|e| NixError::ListGenerations(e.to_string()))?
+		.expect("generation listing panicked")
 		.map_err(|e| NixError::ListGenerations(e.to_string()))
 	}
 }