From 210037fd620ba8f0f70e163a5c203981e9d580ee Mon Sep 17 00:00:00 2001 From: Yaroslav Bolyukin Date: Mon, 15 Jun 2026 01:21:55 +0000 Subject: [PATCH] refactor: nix store api cleanup --- --- 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(()) } --- 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) --- 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 { - 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 { - 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 { + 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 { 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, + pub srcs: Vec, #[serde(default)] - pub drvs: HashMap, + pub drvs: HashMap, } #[derive(Debug, Deserialize)] @@ -90,23 +70,23 @@ #[derive(Debug, Clone)] pub struct DrvGraph { - pub root: String, - pub nodes: HashMap, + pub root: Utf8PathBuf, + pub nodes: HashMap, } #[derive(Debug, Clone)] pub struct DrvNode { pub name: String, - pub input_drvs: HashMap>, - pub input_srcs: Vec, + pub input_drvs: HashMap>, + pub input_srcs: Vec, // TODO: CA outputs without a known paths are skipped - pub outputs: HashMap, + pub outputs: HashMap, } impl DrvGraph { - pub fn resolve(drv_path: &str) -> Result { - let sd = store_dir()?; - let root = to_absolute_store_path(&sd, drv_path); + pub fn resolve(store: &Store, drv_path: &Utf8Path) -> Result { + 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> = parsed + let input_drvs: HashMap> = 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 = parsed + let outputs: HashMap = 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> { - let mut wanted: HashMap> = HashMap::new(); + pub fn wanted_outputs(&self, root_outputs: &[String]) -> HashMap> { + let mut wanted: HashMap> = HashMap::new(); wanted.insert(self.root.clone(), root_outputs.iter().cloned().collect()); - let mut queue: VecDeque = VecDeque::new(); + let mut queue: VecDeque = 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() } --- 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, state: EvalState, } impl GlobalState { fn new() -> Result { 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( - f: impl FnOnce(*mut c_context, *mut c_store, *mut c_eval_state) -> T, -) -> Result { - 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 { - 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> { - 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> { - 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 { - 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 { + GLOBAL_STATE.store.clone() +} + impl Store { pub fn open(uri: &str) -> Result { let uri = CString::new(uri)?; @@ -634,11 +541,108 @@ Ok(Store(ptr)) } - fn parse_path(&self, path: &CStr) -> Result { + pub fn parse_path(&self, path: &Utf8Path) -> Result { + 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 { + 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> { + 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> { + 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 { + 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) { --- 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, + parent: Option, span: Option, refcount: usize, } -static DRV_GRAPH: LazyLock>> = +static DRV_GRAPH: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); -static ACTIVITY_TO_DRV: LazyLock>> = +static ACTIVITY_TO_DRV: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); pub struct BuildGraphGuard { - paths: Vec, + paths: Vec, } impl Drop for BuildGraphGuard { @@ -369,7 +365,7 @@ BuildGraphGuard { paths } } -fn ensure_drv_span(drv_path: &str) -> Option { +fn ensure_drv_span(drv_path: &Utf8Path) -> Option { 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() --- a/crates/nix-eval/src/scheduler.rs +++ b/crates/nix-eval/src/scheduler.rs @@ -1,12 +1,16 @@ use std::collections::{HashMap, HashSet}; +use std::mem; use std::sync::Arc; use anyhow::{Context, Result, bail}; +use camino::{Utf8Path, Utf8PathBuf}; use futures::stream::{FuturesUnordered, StreamExt}; use tokio::sync::{Semaphore, broadcast}; +use tokio::task::spawn_blocking; use tracing::{debug, info, instrument, warn}; use crate::drv::DrvGraph; +use crate::{Store, eval_store}; #[derive(Clone, Debug)] pub enum BuildEvent { @@ -17,31 +21,32 @@ satisfied: usize, }, DrvStarted { - drv_path: String, + drv_path: Utf8PathBuf, name: String, wanted: Vec, }, DrvSkipped { - drv_path: String, + drv_path: Utf8PathBuf, name: String, }, DrvFinished { - drv_path: String, + drv_path: Utf8PathBuf, name: String, }, DrvFailed { - drv_path: String, + drv_path: Utf8PathBuf, name: String, error: String, }, DrvCancelled { - drv_path: String, + drv_path: Utf8PathBuf, name: String, - failed_dep: String, + failed_dep: Utf8PathBuf, }, } pub struct Scheduler { + store: Arc, parallelism: usize, events: broadcast::Sender, } @@ -51,6 +56,7 @@ let parallelism = parallelism.max(1); let (events, _) = broadcast::channel(1024); Self { + store: eval_store(), parallelism, events, } @@ -71,7 +77,7 @@ async fn substitute_prepass( &self, graph: &DrvGraph, - wanted: &HashMap>, + wanted: &HashMap>, ) -> Result<()> { let paths = collect_substitute_paths(graph, wanted); if paths.is_empty() { @@ -82,7 +88,8 @@ .send(BuildEvent::SubstitutePrepassStarted { paths: paths.len() }); debug!("substitute pre-pass: {} paths", paths.len()); - let satisfied = tokio::task::spawn_blocking(move || crate::substitute_paths(&paths)) + let store = self.store.clone(); + let satisfied = spawn_blocking(move || store.substitute_paths(&paths)) .await .expect("substitute pre-pass task should not panic")?; @@ -95,14 +102,14 @@ async fn build_topo( &self, graph: &Arc, - wanted: HashMap>, + wanted: HashMap>, ) -> Result<()> { - let mut indeg: HashMap = graph + let mut indeg: HashMap = graph .nodes .iter() .map(|(k, n)| (k.clone(), n.input_drvs.len())) .collect(); - let mut dependents: HashMap> = HashMap::new(); + let mut dependents: HashMap> = HashMap::new(); for (path, node) in &graph.nodes { for dep in node.input_drvs.keys() { dependents @@ -113,18 +120,18 @@ } let sem = Arc::new(Semaphore::new(self.parallelism)); - let mut ready: Vec = indeg + let mut ready: Vec = indeg .iter() .filter(|(_, d)| **d == 0) .map(|(k, _)| k.clone()) .collect(); let mut in_flight = FuturesUnordered::new(); - let mut failed: HashMap = HashMap::new(); + let mut failed: HashMap = HashMap::new(); // Tainted = transitively depends on a failed drv - let mut tainted: HashMap = HashMap::new(); + let mut tainted: HashMap = HashMap::new(); loop { - let batch: Vec = std::mem::take(&mut ready); + let batch: Vec = mem::take(&mut ready); for path in batch { if let Some(failed_dep) = tainted.get(&path) { let name = graph @@ -145,6 +152,7 @@ let events = self.events.clone(); let graph = graph.clone(); let wanted_here = wanted.get(&path).cloned().unwrap_or_default(); + let store = self.store.clone(); in_flight.push(tokio::spawn(async move { let _permit = sem.acquire_owned().await.expect("semaphore not closed"); let node = graph @@ -158,7 +166,7 @@ && wanted_here.iter().all(|o| { node.outputs .get(o) - .map(|p| crate::is_valid_path(p).unwrap_or(false)) + .map(|p| store.is_valid_path(p)) .unwrap_or(false) }); if all_valid { @@ -176,8 +184,9 @@ }); let path_for_build = path.clone(); - let res = tokio::task::spawn_blocking(move || { - crate::build_drv_outputs(&path_for_build, &wanted_here) + let store = store.clone(); + let res = spawn_blocking(move || { + store.build_drv_outputs(&path_for_build, &wanted_here) }) .await .expect("build task should not panic"); @@ -259,10 +268,10 @@ } fn propagate_done( - dependents: &HashMap>, - indeg: &mut HashMap, - ready: &mut Vec, - finished: &str, + dependents: &HashMap>, + indeg: &mut HashMap, + ready: &mut Vec, + finished: &Utf8Path, ) { if let Some(deps) = dependents.get(finished) { for d in deps { @@ -276,11 +285,11 @@ } fn mark_tainted( - dependents: &HashMap>, - failed: &str, - tainted: &mut HashMap, + dependents: &HashMap>, + failed: &Utf8Path, + tainted: &mut HashMap, ) { - let mut queue: Vec = dependents.get(failed).cloned().unwrap_or_default(); + let mut queue: Vec = dependents.get(failed).cloned().unwrap_or_default(); while let Some(node) = queue.pop() { if tainted .entry(node.clone()) @@ -298,20 +307,17 @@ } } -fn path_to_root(graph: &DrvGraph, from: &str) -> Vec { - let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new(); +fn path_to_root(graph: &DrvGraph, from: &Utf8Path) -> Vec { + let mut dependents: HashMap<&Utf8Path, Vec<&Utf8Path>> = HashMap::new(); for (path, node) in &graph.nodes { for dep in node.input_drvs.keys() { - dependents - .entry(dep.as_str()) - .or_default() - .push(path.as_str()); + dependents.entry(dep).or_default().push(path); } } let mut chain: Vec = vec![node_name(graph, from)]; let mut cur = from; - let mut seen: HashSet<&str> = HashSet::new(); + let mut seen: HashSet<&Utf8Path> = HashSet::new(); seen.insert(cur); while cur != graph.root.as_str() { let Some(next) = dependents.get(cur).and_then(|v| v.first().copied()) else { @@ -326,19 +332,19 @@ chain } -fn node_name(graph: &DrvGraph, path: &str) -> String { +fn node_name(graph: &DrvGraph, path: &Utf8Path) -> String { graph .nodes .get(path) .map(|n| n.name.clone()) - .unwrap_or_else(|| path.to_owned()) + .unwrap_or_else(|| path.to_string()) } fn collect_substitute_paths( graph: &DrvGraph, - wanted: &HashMap>, -) -> Vec { - let mut paths: HashSet = HashSet::new(); + wanted: &HashMap>, +) -> Vec { + let mut paths: HashSet = HashSet::new(); for node in graph.nodes.values() { for src in &node.input_srcs { paths.insert(src.clone()); --- 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, 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())) } } -- gitstuff