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

24#[cfg(feature = "indicatif")]24#[cfg(feature = "indicatif")]
25use indicatif::{ProgressState, ProgressStyle};25use indicatif::{ProgressState, ProgressStyle};
26use nix_eval::{26use nix_eval::{
27	add_file_to_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries,27	eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
28	init_tokio_for_nix,
29};28};
30use opentelemetry::trace::TracerProvider;29use opentelemetry::trace::TracerProvider;
33	OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,32	OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,
34};33};
35use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};34use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};
35use tokio::task::spawn_blocking;
36use tracing::{Instrument, error, info, info_span};36use tracing::{Instrument, error, info, info_span};
37#[cfg(feature = "indicatif")]37#[cfg(feature = "indicatif")]
38use tracing_indicatif::IndicatifLayer;38use tracing_indicatif::IndicatifLayer;
59				Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;59				Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;
60			let span = info_span!("prefetching", name = %name);60			let span = info_span!("prefetching", name = %name);
61			tasks.push(async move {61			tasks.push(async move {
62				let store = eval_store();
62				let added = tokio::task::spawn_blocking(move || add_file_to_store(&name, &path))63				let added = spawn_blocking(move || store.add_file(&name, &path))
63					.instrument(span.clone())64					.instrument(span.clone())
64					.await??;65					.await??;
65				let _g = span.enter();66				let _g = span.enter();
121		Opts::Prefetch(p) => p.run(config).await?,122		Opts::Prefetch(p) => p.run(config).await?,
122		Opts::Tf(t) => t.run(config).await?,123		Opts::Tf(t) => t.run(config).await?,
123		// TODO: actually parse commands before starting the async runtime124		// TODO: actually parse commands before starting the async runtime
124		Opts::Complete(c) => {125		Opts::Complete(c) => spawn_blocking(move || c.run(RootOpts::command())).await?,
125			tokio::task::spawn_blocking(move || c.run(RootOpts::command())).await?
126		}
127	};126	};
128	Ok(())127	Ok(())
129}128}

13use camino::{Utf8Path, Utf8PathBuf};13use camino::{Utf8Path, Utf8PathBuf};
14use chrono::{DateTime, Utc};14use chrono::{DateTime, Utc};
15use fleet_shared::SecretData;15use fleet_shared::SecretData;
16use nix_eval::{Store, Value, nix_go, nix_go_json, util::assert_warn};16use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};
17use remowt_client::{AgentBundle, Remowt};17use remowt_client::{AgentBundle, Remowt};
18use remowt_endpoints::fs::FsClient;18use remowt_endpoints::fs::FsClient;
19use remowt_link_shared::Address;19use remowt_link_shared::Address;
427		let store = self.nix_store().await?;427		let store = self.nix_store().await?;
428		{428		{
429			let path = path.clone();429			let path = path.clone();
430			let store = eval_store();
430			spawn_blocking(move || nix_eval::copy_closure_to(&store, path.as_ref()))431			spawn_blocking(move || store.copy_to(&store, path.as_ref()))
431				.await?432				.await
433				.expect("copy_to panicked")
432				.context("copying closure to remote store")?;434				.context("copying closure to remote store")?;
433		}435		}
434		Ok(path)436		Ok(path)

1use std::collections::{HashMap, HashSet, VecDeque};1use std::collections::{HashMap, HashSet, VecDeque};
2
2use std::ffi::CString;3use anyhow::{Result, bail};
3
4use anyhow::{Result, bail};4use camino::{Utf8Component, Utf8Path, Utf8PathBuf};
5use serde::Deserialize;5use serde::Deserialize;
66
7use crate::nix_raw::{derivation_free, derivation_to_json, store_drv_from_store_path};7use crate::nix_raw::{derivation_free, derivation_to_json, store_drv_from_store_path};
8use crate::{copy_nix_str, with_store_context};8use crate::{Store, copy_nix_str, with_default_context};
9
10fn store_dir() -> Result<String> {
11	let mut out = String::new();
12	with_store_context(|c, store, _| unsafe {
13		crate::nix_raw::store_get_storedir(c, store, Some(copy_nix_str), (&raw mut out).cast())
14	})?;
15	Ok(out)
16}
17
18fn to_absolute_store_path(store_dir: &str, path: &str) -> String {
19	if path.starts_with('/') {
20		path.to_owned()
21	} else {
22		format!("{store_dir}/{path}")
23	}
24}
259
26pub struct Derivation(*mut crate::nix_raw::derivation);10pub struct Derivation(*mut crate::nix_raw::derivation);
27unsafe impl Send for Derivation {}11unsafe impl Send for Derivation {}
2812
29impl Derivation {13impl Derivation {
30	pub fn from_path(drv_path: &str) -> Result<Self> {14	pub fn from_path(store: &Store, drv_path: &Utf8Path) -> Result<Self> {
31		let path_c = CString::new(drv_path)?;15		let store_path = store.parse_path(drv_path)?;
32		let store_path = with_store_context(|c, store, _| unsafe {16		let drv = with_default_context(|c, _| unsafe {
33			crate::nix_raw::store_parse_path(c, store, path_c.as_ptr())17			store_drv_from_store_path(c, store.as_ptr(), store_path.as_ptr())
34		})?;18		});
35		let drv = with_store_context(|c, store, _| unsafe {
36			store_drv_from_store_path(c, store, store_path)
37		});
38		unsafe { crate::nix_raw::store_path_free(store_path) };
39		let drv = drv?;19		let drv = drv?;
40		if drv.is_null() {20		if drv.is_null() {
41			bail!("failed to read derivation from {drv_path}");21			bail!("failed to read derivation from {drv_path}");
4525
46	pub fn to_json_string(&self) -> Result<String> {26	pub fn to_json_string(&self) -> Result<String> {
47		let mut out = String::new();27		let mut out = String::new();
48		with_store_context(|c, _, _| unsafe {28		with_default_context(|c, _| unsafe {
49			derivation_to_json(c, self.0, Some(copy_nix_str), (&raw mut out).cast())29			derivation_to_json(c, self.0, Some(copy_nix_str), (&raw mut out).cast())
50		})?;30		})?;
51		Ok(out)31		Ok(out)
78#[derive(Debug, Deserialize)]58#[derive(Debug, Deserialize)]
79pub struct DrvInputs {59pub struct DrvInputs {
80	#[serde(default)]60	#[serde(default)]
81	pub srcs: Vec<String>,61	pub srcs: Vec<Utf8PathBuf>,
82	#[serde(default)]62	#[serde(default)]
83	pub drvs: HashMap<String, DrvInputEntry>,63	pub drvs: HashMap<Utf8PathBuf, DrvInputEntry>,
84}64}
8565
86#[derive(Debug, Deserialize)]66#[derive(Debug, Deserialize)]
9070
91#[derive(Debug, Clone)]71#[derive(Debug, Clone)]
92pub struct DrvGraph {72pub struct DrvGraph {
93	pub root: String,73	pub root: Utf8PathBuf,
94	pub nodes: HashMap<String, DrvNode>,74	pub nodes: HashMap<Utf8PathBuf, DrvNode>,
95}75}
9676
97#[derive(Debug, Clone)]77#[derive(Debug, Clone)]
98pub struct DrvNode {78pub struct DrvNode {
99	pub name: String,79	pub name: String,
100	pub input_drvs: HashMap<String, Vec<String>>,80	pub input_drvs: HashMap<Utf8PathBuf, Vec<String>>,
101	pub input_srcs: Vec<String>,81	pub input_srcs: Vec<Utf8PathBuf>,
102	// TODO: CA outputs without a known paths are skipped82	// TODO: CA outputs without a known paths are skipped
103	pub outputs: HashMap<String, String>,83	pub outputs: HashMap<String, Utf8PathBuf>,
104}84}
10585
106impl DrvGraph {86impl DrvGraph {
107	pub fn resolve(drv_path: &str) -> Result<Self> {87	pub fn resolve(store: &Store, drv_path: &Utf8Path) -> Result<Self> {
108		let sd = store_dir()?;88		let sd = store.store_dir()?;
109		let root = to_absolute_store_path(&sd, drv_path);89		let root = sd.join(drv_path);
11090
111		let mut nodes = HashMap::new();91		let mut nodes = HashMap::new();
112		let mut queue = VecDeque::new();92		let mut queue = VecDeque::new();
115		visited.insert(root.clone());95		visited.insert(root.clone());
11696
117		while let Some(path) = queue.pop_front() {97		while let Some(path) = queue.pop_front() {
118			let drv = Derivation::from_path(&path)?;98			let drv = Derivation::from_path(store, &path)?;
119			let parsed = drv.parsed()?;99			let parsed = drv.parsed()?;
120100
121			let input_drvs: HashMap<String, Vec<String>> = parsed101			let input_drvs: HashMap<Utf8PathBuf, Vec<String>> = parsed
122				.inputs102				.inputs
123				.drvs103				.drvs
124				.into_iter()104				.into_iter()
125				.map(|(k, v)| (to_absolute_store_path(&sd, &k), v.outputs))105				.map(|(k, v)| (sd.join(&k), v.outputs))
126				.collect();106				.collect();
127107
128			for dep_path in input_drvs.keys() {108			for dep_path in input_drvs.keys() {
131				}111				}
132			}112			}
133113
134			let outputs: HashMap<String, String> = parsed114			let outputs: HashMap<String, Utf8PathBuf> = parsed
135				.outputs115				.outputs
136				.into_iter()116				.into_iter()
137				.filter_map(|(name, out)| out.path.map(|p| (name, to_absolute_store_path(&sd, &p))))117				.filter_map(|(name, out)| out.path.map(|p| (name, sd.join(&p))))
138				.collect();118				.collect();
139119
140			nodes.insert(120			nodes.insert(
151		Ok(Self { root, nodes })131		Ok(Self { root, nodes })
152	}132	}
153133
154	pub fn wanted_outputs(&self, root_outputs: &[String]) -> HashMap<String, Vec<String>> {134	pub fn wanted_outputs(&self, root_outputs: &[String]) -> HashMap<Utf8PathBuf, Vec<String>> {
155		let mut wanted: HashMap<String, HashSet<String>> = HashMap::new();135		let mut wanted: HashMap<Utf8PathBuf, HashSet<String>> = HashMap::new();
156		wanted.insert(self.root.clone(), root_outputs.iter().cloned().collect());136		wanted.insert(self.root.clone(), root_outputs.iter().cloned().collect());
157137
158		let mut queue: VecDeque<String> = VecDeque::new();138		let mut queue: VecDeque<Utf8PathBuf> = VecDeque::new();
159		queue.push_back(self.root.clone());139		queue.push_back(self.root.clone());
160		while let Some(path) = queue.pop_front() {140		while let Some(path) = queue.pop_front() {
161			let Some(node) = self.nodes.get(&path) else {141			let Some(node) = self.nodes.get(&path) else {
186	}166	}
187}167}
188168
189fn extract_drv_name(drv_path: &str) -> String {169pub fn extract_drv_name(drv_path: &Utf8Path) -> String {
190	drv_path170	let comp = drv_path
191		.rsplit('/')
192		.next()171		.components()
193		.and_then(|f| f.strip_suffix(".drv"))172		.rev()
194		.and_then(|f| f.split_once('-').map(|(_, name)| name))173		.next()
195		.unwrap_or(drv_path)174		.expect("drv path is at least one component");
196		.to_owned()175	let Utf8Component::Normal(n) = comp else {
176		panic!("drv path is normal");
177	};
178
179	let n = n.strip_suffix(".drv").unwrap_or(n);
180
181	let n = n.split_once(' ').map(|(_, n)| n).unwrap_or(n);
182
183	n.to_owned()
197}184}
198185

25	PrimOpFun, Store as c_store, StorePath as c_store_path, alloc_primop, alloc_value,25	PrimOpFun, Store as c_store, StorePath as c_store_path, alloc_primop, alloc_value,
26	bindings_builder_free, bindings_builder_insert, c_context, c_context_create, c_context_free,26	bindings_builder_free, bindings_builder_insert, c_context, c_context_create, c_context_free,
27	clear_err, copy_value, err_NIX_ERR_KEY, err_NIX_ERR_NIX_ERROR, err_NIX_ERR_OVERFLOW,27	clear_err, copy_value, err_NIX_ERR_KEY, err_NIX_ERR_NIX_ERROR, err_NIX_ERR_OVERFLOW,
28	err_NIX_ERR_UNKNOWN, err_code, err_info_msg, err_msg, eval_state_build,28	err_NIX_ERR_UNKNOWN, err_NIX_OK, err_code, err_info_msg, err_msg, eval_state_build,
29	eval_state_builder_load, eval_state_builder_new, eval_state_builder_set_eval_setting,29	eval_state_builder_load, eval_state_builder_new, eval_state_builder_set_eval_setting,
30	expr_eval_from_string, fetchers_settings, fetchers_settings_free, fetchers_settings_new,30	expr_eval_from_string, fetchers_settings, fetchers_settings_free, fetchers_settings_new,
31	flake_lock, flake_lock_flags, flake_lock_flags_free, flake_lock_flags_new, flake_reference,31	flake_lock, flake_lock_flags, flake_lock_flags_free, flake_lock_flags_new, flake_reference,
320struct GlobalState {320struct GlobalState {
321	// Store should be valid as long as EvalState is valid321	// Store should be valid as long as EvalState is valid
322	#[allow(dead_code)]322	#[allow(dead_code)]
323	store: Store,323	store: Arc<Store>,
324	state: EvalState,324	state: EvalState,
325}325}
326impl GlobalState {326impl GlobalState {
327	fn new() -> Result<Self> {327	fn new() -> Result<Self> {
328		let mut ctx = NixContext::new();328		let mut ctx = NixContext::new();
329		let store = ctx329		let store = Arc::new(
330			.run_in_context(|c| unsafe { store_open(c, c"auto".as_ptr(), null_mut()) })330			ctx.run_in_context(|c| unsafe { store_open(c, c"auto".as_ptr(), null_mut()) })
331			.map(Store)?;331				.map(Store)?,
332		);
332333
333		let builder = ctx.run_in_context(|c| unsafe { eval_state_builder_new(c, store.0) })?;334		let builder = ctx.run_in_context(|c| unsafe { eval_state_builder_new(c, store.0) })?;
334		ctx.run_in_context(|c| unsafe { eval_state_builder_load(c, builder) })?;335		ctx.run_in_context(|c| unsafe { eval_state_builder_load(c, builder) })?;
385	v386	v
386}387}
387
388/// Same as with_default_context, but also passes store...
389/// Yep, this code is garbage and needs to be refactored.
390pub(crate) fn with_store_context<T>(
391	f: impl FnOnce(*mut c_context, *mut c_store, *mut c_eval_state) -> T,
392) -> Result<T> {
393	let global = &GLOBAL_STATE;
394	let (ctx, store, state) =
395		THREAD_STATE.with_borrow_mut(|w| (w.ctx.0, global.store.0, global.state.0));
396	let mut ctx = NixContext(ctx);
397	let v = ctx.run_in_context(|c| f(c, store, state));
398	std::mem::forget(ctx);
399	v
400}
401388
402pub fn set_setting(s: &CStr, v: &CStr) -> Result<()> {389pub fn set_setting(s: &CStr, v: &CStr) -> Result<()> {
403	with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())390	with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())
404}391}
405
406#[instrument(skip(dst))]
407pub fn copy_closure_to(dst: &Store, path: &Utf8Path) -> Result<()> {
408	let path_c = CString::new(path.as_str())?;
409	with_store_context(|c, src_store, _state| -> Result<()> {
410		let sp = unsafe { store_parse_path(c, src_store, path_c.as_ptr()) };
411		if sp.is_null() {
412			bail!("failed to parse store path {path}");
413		}
414		let rc = unsafe { store_copy_closure(c, src_store, dst.0, sp) };
415		unsafe { store_path_free(sp) };
416		if rc != nix_raw::err_NIX_OK {
417			bail!("store_copy_closure failed (code {rc})");
418		}
419		Ok(())
420	})?
421}
422
423#[instrument]
424pub fn switch_profile(profile: &str, store_path: &Utf8Path) -> Result<()> {
425	let msg = with_store_context(|_c, store, _state| unsafe {
426		nix_cxx::switch_profile(store.cast(), profile, store_path.as_str())
427	})?
428	.to_string();
429	if msg.is_empty() {
430		Ok(())
431	} else {
432		bail!("failed to switch profile {profile}: {msg}");
433	}
434}
435
436// TODO: fleet operator-managed key file
437#[instrument]
438pub fn sign_closure(store_path: &str, key_file: &str) -> Result<()> {
439	let msg = with_store_context(|_c, store, _state| unsafe {
440		nix_cxx::sign_closure(store.cast(), store_path, key_file)
441	})?
442	.to_string();
443	if msg.is_empty() {
444		Ok(())
445	} else {
446		bail!("failed to sign {store_path}: {msg}");
447	}
448}
449392
450#[derive(Debug)]393#[derive(Debug)]
451pub struct AddedFile {394pub struct AddedFile {
482		.collect())425		.collect())
483}426}
484
485#[instrument]
486pub fn add_file_to_store(name: &str, path: &Utf8Path) -> Result<AddedFile> {
487	let res = with_store_context(|_c, store, _state| unsafe {
488		nix_cxx::add_file_to_store(store.cast(), name, path.as_str())
489	})?;
490	if !res.error.is_empty() {
491		bail!("failed to add {path} to store: {}", res.error);
492	}
493	Ok(AddedFile {
494		store_path: Utf8PathBuf::from(res.store_path),
495		hash: res.hash,
496	})
497}
498
499pub fn build_drv_outputs(drv_path: &str, output_names: &[String]) -> Result<Vec<String>> {
500	let joined = output_names.join("\n");
501	let res = with_store_context(|_c, store, _state| unsafe {
502		nix_cxx::build_drv_outputs(store.cast(), drv_path, &joined)
503	})?;
504	if !res.error.is_empty() {
505		bail!("build of {drv_path} failed: {}", res.error);
506	}
507	Ok(res.outputs)
508}
509
510pub fn substitute_paths(paths: &[String]) -> Result<Vec<String>> {
511	let joined = paths.join("\n");
512	let res = with_store_context(|_c, store, _state| unsafe {
513		nix_cxx::substitute_paths(store.cast(), &joined)
514	})?;
515	if !res.error.is_empty() {
516		warn!("substitute_paths reported: {}", res.error);
517	}
518	Ok(res.outputs)
519}
520
521pub fn is_valid_path(path: &str) -> Result<bool> {
522	with_store_context(|_c, store, _state| unsafe { nix_cxx::is_valid_path(store.cast(), path) })
523}
524427
525pub struct FetchSettings(*mut fetchers_settings);428pub struct FetchSettings(*mut fetchers_settings);
526impl FetchSettings {429impl FetchSettings {
624unsafe impl Send for Store {}527unsafe impl Send for Store {}
625unsafe impl Sync for Store {}528unsafe impl Sync for Store {}
529
530pub fn eval_store() -> Arc<Store> {
531	GLOBAL_STATE.store.clone()
532}
626533
627impl Store {534impl Store {
628	pub fn open(uri: &str) -> Result<Self> {535	pub fn open(uri: &str) -> Result<Self> {
634		Ok(Store(ptr))541		Ok(Store(ptr))
635	}542	}
636543
637	fn parse_path(&self, path: &CStr) -> Result<StorePath> {544	pub fn parse_path(&self, path: &Utf8Path) -> Result<StorePath> {
545		let path = CString::new(path.as_str()).expect("valid cstr");
638		with_default_context(|c, _| {546		with_default_context(|c, _| {
639			StorePath(unsafe { store_parse_path(c, self.0, path.as_ptr()) })547			StorePath(unsafe { store_parse_path(c, self.0, path.as_ptr()) })
640		})548		})
641	}549	}
550
551	#[instrument(skip(self))]
552	pub fn sign_closure(&self, path: &Utf8Path, key_file: &Utf8Path) -> Result<()> {
553		let err = with_default_context(|_, _| unsafe {
554			nix_cxx::sign_closure(self.as_ptr().cast(), path.as_str(), key_file.as_str())
555		})?
556		.to_string();
557
558		if err.is_empty() {
559			Ok(())
560		} else {
561			bail!("failed to sign {path}: {err}");
562		}
563	}
564
565	#[instrument(skip(self, dst))]
566	pub fn copy_to(&self, dst: &Store, path: &Utf8Path) -> Result<()> {
567		let sp = self
568			.parse_path(&path)
569			.context("failed to parse store path")?;
570		let rc = with_default_context(|c, _| unsafe {
571			store_copy_closure(c, self.as_ptr(), dst.0, sp.as_ptr())
572		})?;
573		if rc != err_NIX_OK {
574			bail!("store_copy_closure failed (code {rc})");
575		}
576		Ok(())
577	}
578
579	/// Would only work with local store.
580	#[instrument(skip(self))]
581	pub fn switch_profile(&self, profile: &str, path: &Utf8Path) -> Result<()> {
582		let msg = unsafe { nix_cxx::switch_profile(self.as_ptr().cast(), profile, path.as_str()) };
583		if msg.is_empty() {
584			Ok(())
585		} else {
586			bail!("failed to switch profile {profile}: {msg}");
587		}
588	}
589
590	#[instrument(skip(self))]
591	pub fn add_file(&self, name: &str, path: &Utf8Path) -> Result<AddedFile> {
592		let msg = unsafe { nix_cxx::add_file_to_store(self.as_ptr().cast(), name, path.as_str()) };
593		if !msg.error.is_empty() {
594			bail!("failed to add {path} to store: {}", msg.error)
595		}
596		Ok(AddedFile {
597			store_path: Utf8PathBuf::from(msg.store_path),
598			hash: msg.hash,
599		})
600	}
601
602	#[instrument(skip(self))]
603	pub fn substitute_paths(&self, paths: &[Utf8PathBuf]) -> Result<Vec<Utf8PathBuf>> {
604		let joined = paths.into_iter().join("\n");
605		let res = unsafe { nix_cxx::substitute_paths(self.as_ptr().cast(), &joined) };
606		if !res.error.is_empty() {
607			warn!("substitute_paths reported: {}", res.error);
608		}
609		Ok(res.outputs.into_iter().map(Utf8PathBuf::from).collect())
610	}
611
612	#[instrument(skip(self))]
613	pub fn is_valid_path(&self, path: &Utf8Path) -> bool {
614		unsafe { nix_cxx::is_valid_path(self.as_ptr().cast(), path.as_str()) }
615	}
616
617	#[instrument(skip(self))]
618	pub fn build_drv_outputs(
619		&self,
620		drv_path: &Utf8Path,
621		output_names: &[String],
622	) -> Result<Vec<String>> {
623		let joined = output_names.join("\n");
624		let res =
625			unsafe { nix_cxx::build_drv_outputs(self.as_ptr().cast(), drv_path.as_str(), &joined) };
626		if !res.error.is_empty() {
627			bail!("build of {drv_path} failed: {}", res.error);
628		}
629		Ok(res.outputs)
630	}
631
632	#[instrument(skip(self))]
633	pub fn store_dir(&self) -> Result<Utf8PathBuf> {
634		let mut out = String::new();
635		with_default_context(|c, es| unsafe {
636			nix_raw::store_get_storedir(c, self.as_ptr(), Some(copy_nix_str), (&raw mut out).cast())
637		})?;
638		let p = Utf8PathBuf::from(out);
639		assert!(p.is_absolute());
640		Ok(p)
641	}
642
643	fn as_ptr(&self) -> *mut c_store {
644		self.0
645	}
642}646}
643impl Drop for Store {647impl Drop for Store {
644	fn drop(&mut self) {648	fn drop(&mut self) {
1060			self.clone()1064			self.clone()
1061		};1065		};
10621066
1063		let drv_path = v1067		let drv_path = Utf8PathBuf::from(
1064			.get_field("drvPath")1068			v.get_field("drvPath")
1065			.context("getting drvPath")?1069				.context("getting drvPath")?
1066			.to_string()?;1070				.to_string()?,
1071		);
1067		let graph = Arc::new(drv::DrvGraph::resolve(&drv_path)?);1072		let graph = Arc::new(drv::DrvGraph::resolve(&eval_store(), &drv_path)?);
1068		let _guard = logging::register_build_graph(&Span::current(), &graph);1073		let _guard = logging::register_build_graph(&Span::current(), &graph);
10691074
1070		scheduler::build_graph_sync(graph.clone(), vec![output.to_owned()])?;1075		scheduler::build_graph_sync(graph.clone(), vec![output.to_owned()])?;
1255	}1260	}
1256}1261}
12571262
1258struct StorePath(*mut c_store_path);1263pub struct StorePath(*mut c_store_path);
1259impl StorePath {}1264impl StorePath {
1265	fn as_ptr(&self) -> *mut c_store_path {
1266		self.0
1267	}
1268}
12601269
1261impl Drop for StorePath {1270impl Drop for StorePath {

2use std::fmt::Arguments;2use std::fmt::Arguments;
3use std::sync::{LazyLock, Mutex};3use std::sync::{LazyLock, Mutex};
44
5use camino::{Utf8Path, Utf8PathBuf};
5use cxx::ExternType;6use cxx::ExternType;
6use tracing::{7use tracing::{
7	Level, Span, debug, debug_span, error, error_span, info, info_span, trace, trace_span, warn,8	Level, Span, debug, debug_span, error, error_span, info, info_span, trace, trace_span, warn,
11use tracing_indicatif::span_ext::IndicatifSpanExt as _;12use tracing_indicatif::span_ext::IndicatifSpanExt as _;
12use vte::Parser;13use vte::Parser;
14
15use crate::drv::extract_drv_name;
1316
14#[derive(Debug)]17#[derive(Debug)]
15enum ActivityType {18enum ActivityType {
33	a.strip_prefix(pref)?.strip_suffix(suff)36	a.strip_prefix(pref)?.strip_suffix(suff)
34}37}
3538
36fn parse_path(path: &str) -> &str {39fn parse_path(path: &str) -> Utf8PathBuf {
37	strip_prefix_suffix(path, "\x1b[35;1m", "\x1b[0m").unwrap_or(path)40	Utf8PathBuf::from(strip_prefix_suffix(path, "\x1b[35;1m", "\x1b[0m").unwrap_or(path))
38}41}
3942
40fn parse_drv(drv: &str) -> &str {43fn parse_drv(drv: &str) -> String {
41	let drv = parse_path(drv);44	let drv = parse_path(drv);
42	if let Some(pkg) = drv.strip_prefix("/nix/store/") {45	extract_drv_name(&drv)
43		let mut it = pkg.splitn(2, '-');
44		it.next();
45		if let Some(pkg) = it.next() {
46			return pkg;
47		}
48	}
49	drv
50}46}
51fn parse_host(host: &str) -> &str {47fn parse_host(host: &str) -> &str {
52	if host.is_empty() || host == "local" {48	if host.is_empty() || host == "local" {
287283
288struct DrvGraphEntry {284struct DrvGraphEntry {
289	name: String,285	name: String,
290	parent: Option<String>,286	parent: Option<Utf8PathBuf>,
291	span: Option<Span>,287	span: Option<Span>,
292	refcount: usize,288	refcount: usize,
293}289}
294290
295static DRV_GRAPH: LazyLock<Mutex<HashMap<String, DrvGraphEntry>>> =291static DRV_GRAPH: LazyLock<Mutex<HashMap<Utf8PathBuf, DrvGraphEntry>>> =
296	LazyLock::new(|| Mutex::new(HashMap::new()));292	LazyLock::new(|| Mutex::new(HashMap::new()));
297293
298static ACTIVITY_TO_DRV: LazyLock<Mutex<HashMap<u64, String>>> =294static ACTIVITY_TO_DRV: LazyLock<Mutex<HashMap<u64, Utf8PathBuf>>> =
299	LazyLock::new(|| Mutex::new(HashMap::new()));295	LazyLock::new(|| Mutex::new(HashMap::new()));
300296
301pub struct BuildGraphGuard {297pub struct BuildGraphGuard {
302	paths: Vec<String>,298	paths: Vec<Utf8PathBuf>,
303}299}
304300
305impl Drop for BuildGraphGuard {301impl Drop for BuildGraphGuard {
369	BuildGraphGuard { paths }365	BuildGraphGuard { paths }
370}366}
371367
372fn ensure_drv_span(drv_path: &str) -> Option<Span> {368fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {
373	let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");369	let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");
374370
375	if let Some(span) = drv_graph.get(drv_path).and_then(|e| e.span.clone()) {371	if let Some(span) = drv_graph.get(drv_path).and_then(|e| e.span.clone()) {
442			self.fields.first().and_then(|f| match f {438			self.fields.first().and_then(|f| match f {
443				FieldValue::Str(drv_path) => {439				FieldValue::Str(drv_path) => {
444					let clean = parse_path(drv_path);440					let clean = parse_path(drv_path);
445					let span = ensure_drv_span(clean);441					let span = ensure_drv_span(&clean);
446					if span.is_some() {442					if span.is_some() {
447						ACTIVITY_TO_DRV443						ACTIVITY_TO_DRV
448							.lock()444							.lock()

1use std::collections::{HashMap, HashSet};1use std::collections::{HashMap, HashSet};
2use std::mem;
2use std::sync::Arc;3use std::sync::Arc;
34
4use anyhow::{Context, Result, bail};5use anyhow::{Context, Result, bail};
6use camino::{Utf8Path, Utf8PathBuf};
5use futures::stream::{FuturesUnordered, StreamExt};7use futures::stream::{FuturesUnordered, StreamExt};
6use tokio::sync::{Semaphore, broadcast};8use tokio::sync::{Semaphore, broadcast};
9use tokio::task::spawn_blocking;
7use tracing::{debug, info, instrument, warn};10use tracing::{debug, info, instrument, warn};
811
9use crate::drv::DrvGraph;12use crate::drv::DrvGraph;
13use crate::{Store, eval_store};
1014
11#[derive(Clone, Debug)]15#[derive(Clone, Debug)]
12pub enum BuildEvent {16pub enum BuildEvent {
17		satisfied: usize,21		satisfied: usize,
18	},22	},
19	DrvStarted {23	DrvStarted {
20		drv_path: String,24		drv_path: Utf8PathBuf,
21		name: String,25		name: String,
22		wanted: Vec<String>,26		wanted: Vec<String>,
23	},27	},
24	DrvSkipped {28	DrvSkipped {
25		drv_path: String,29		drv_path: Utf8PathBuf,
26		name: String,30		name: String,
27	},31	},
28	DrvFinished {32	DrvFinished {
29		drv_path: String,33		drv_path: Utf8PathBuf,
30		name: String,34		name: String,
31	},35	},
32	DrvFailed {36	DrvFailed {
33		drv_path: String,37		drv_path: Utf8PathBuf,
34		name: String,38		name: String,
35		error: String,39		error: String,
36	},40	},
37	DrvCancelled {41	DrvCancelled {
38		drv_path: String,42		drv_path: Utf8PathBuf,
39		name: String,43		name: String,
40		failed_dep: String,44		failed_dep: Utf8PathBuf,
41	},45	},
42}46}
4347
44pub struct Scheduler {48pub struct Scheduler {
49	store: Arc<Store>,
45	parallelism: usize,50	parallelism: usize,
46	events: broadcast::Sender<BuildEvent>,51	events: broadcast::Sender<BuildEvent>,
47}52}
51		let parallelism = parallelism.max(1);56		let parallelism = parallelism.max(1);
52		let (events, _) = broadcast::channel(1024);57		let (events, _) = broadcast::channel(1024);
53		Self {58		Self {
59			store: eval_store(),
54			parallelism,60			parallelism,
55			events,61			events,
56		}62		}
71	async fn substitute_prepass(77	async fn substitute_prepass(
72		&self,78		&self,
73		graph: &DrvGraph,79		graph: &DrvGraph,
74		wanted: &HashMap<String, Vec<String>>,80		wanted: &HashMap<Utf8PathBuf, Vec<String>>,
75	) -> Result<()> {81	) -> Result<()> {
76		let paths = collect_substitute_paths(graph, wanted);82		let paths = collect_substitute_paths(graph, wanted);
77		if paths.is_empty() {83		if paths.is_empty() {
82			.send(BuildEvent::SubstitutePrepassStarted { paths: paths.len() });88			.send(BuildEvent::SubstitutePrepassStarted { paths: paths.len() });
83		debug!("substitute pre-pass: {} paths", paths.len());89		debug!("substitute pre-pass: {} paths", paths.len());
8490
91		let store = self.store.clone();
85		let satisfied = tokio::task::spawn_blocking(move || crate::substitute_paths(&paths))92		let satisfied = spawn_blocking(move || store.substitute_paths(&paths))
86			.await93			.await
87			.expect("substitute pre-pass task should not panic")?;94			.expect("substitute pre-pass task should not panic")?;
8895
95	async fn build_topo(102	async fn build_topo(
96		&self,103		&self,
97		graph: &Arc<DrvGraph>,104		graph: &Arc<DrvGraph>,
98		wanted: HashMap<String, Vec<String>>,105		wanted: HashMap<Utf8PathBuf, Vec<String>>,
99	) -> Result<()> {106	) -> Result<()> {
100		let mut indeg: HashMap<String, usize> = graph107		let mut indeg: HashMap<Utf8PathBuf, usize> = graph
101			.nodes108			.nodes
102			.iter()109			.iter()
103			.map(|(k, n)| (k.clone(), n.input_drvs.len()))110			.map(|(k, n)| (k.clone(), n.input_drvs.len()))
104			.collect();111			.collect();
105		let mut dependents: HashMap<String, Vec<String>> = HashMap::new();112		let mut dependents: HashMap<Utf8PathBuf, Vec<Utf8PathBuf>> = HashMap::new();
106		for (path, node) in &graph.nodes {113		for (path, node) in &graph.nodes {
107			for dep in node.input_drvs.keys() {114			for dep in node.input_drvs.keys() {
108				dependents115				dependents
113		}120		}
114121
115		let sem = Arc::new(Semaphore::new(self.parallelism));122		let sem = Arc::new(Semaphore::new(self.parallelism));
116		let mut ready: Vec<String> = indeg123		let mut ready: Vec<Utf8PathBuf> = indeg
117			.iter()124			.iter()
118			.filter(|(_, d)| **d == 0)125			.filter(|(_, d)| **d == 0)
119			.map(|(k, _)| k.clone())126			.map(|(k, _)| k.clone())
120			.collect();127			.collect();
121		let mut in_flight = FuturesUnordered::new();128		let mut in_flight = FuturesUnordered::new();
122		let mut failed: HashMap<String, String> = HashMap::new();129		let mut failed: HashMap<Utf8PathBuf, String> = HashMap::new();
123		// Tainted = transitively depends on a failed drv130		// Tainted = transitively depends on a failed drv
124		let mut tainted: HashMap<String, String> = HashMap::new();131		let mut tainted: HashMap<Utf8PathBuf, Utf8PathBuf> = HashMap::new();
125132
126		loop {133		loop {
127			let batch: Vec<String> = std::mem::take(&mut ready);134			let batch: Vec<Utf8PathBuf> = mem::take(&mut ready);
128			for path in batch {135			for path in batch {
129				if let Some(failed_dep) = tainted.get(&path) {136				if let Some(failed_dep) = tainted.get(&path) {
130					let name = graph137					let name = graph
145				let events = self.events.clone();152				let events = self.events.clone();
146				let graph = graph.clone();153				let graph = graph.clone();
147				let wanted_here = wanted.get(&path).cloned().unwrap_or_default();154				let wanted_here = wanted.get(&path).cloned().unwrap_or_default();
155				let store = self.store.clone();
148				in_flight.push(tokio::spawn(async move {156				in_flight.push(tokio::spawn(async move {
149					let _permit = sem.acquire_owned().await.expect("semaphore not closed");157					let _permit = sem.acquire_owned().await.expect("semaphore not closed");
150					let node = graph158					let node = graph
158						&& wanted_here.iter().all(|o| {166						&& wanted_here.iter().all(|o| {
159							node.outputs167							node.outputs
160								.get(o)168								.get(o)
161								.map(|p| crate::is_valid_path(p).unwrap_or(false))169								.map(|p| store.is_valid_path(p))
162								.unwrap_or(false)170								.unwrap_or(false)
163						});171						});
164					if all_valid {172					if all_valid {
176					});184					});
177185
178					let path_for_build = path.clone();186					let path_for_build = path.clone();
187					let store = store.clone();
179					let res = tokio::task::spawn_blocking(move || {188					let res = spawn_blocking(move || {
180						crate::build_drv_outputs(&path_for_build, &wanted_here)189						store.build_drv_outputs(&path_for_build, &wanted_here)
181					})190					})
182					.await191					.await
183					.expect("build task should not panic");192					.expect("build task should not panic");
259}268}
260269
261fn propagate_done(270fn propagate_done(
262	dependents: &HashMap<String, Vec<String>>,271	dependents: &HashMap<Utf8PathBuf, Vec<Utf8PathBuf>>,
263	indeg: &mut HashMap<String, usize>,272	indeg: &mut HashMap<Utf8PathBuf, usize>,
264	ready: &mut Vec<String>,273	ready: &mut Vec<Utf8PathBuf>,
265	finished: &str,274	finished: &Utf8Path,
266) {275) {
267	if let Some(deps) = dependents.get(finished) {276	if let Some(deps) = dependents.get(finished) {
268		for d in deps {277		for d in deps {
276}285}
277286
278fn mark_tainted(287fn mark_tainted(
279	dependents: &HashMap<String, Vec<String>>,288	dependents: &HashMap<Utf8PathBuf, Vec<Utf8PathBuf>>,
280	failed: &str,289	failed: &Utf8Path,
281	tainted: &mut HashMap<String, String>,290	tainted: &mut HashMap<Utf8PathBuf, Utf8PathBuf>,
282) {291) {
283	let mut queue: Vec<String> = dependents.get(failed).cloned().unwrap_or_default();292	let mut queue: Vec<Utf8PathBuf> = dependents.get(failed).cloned().unwrap_or_default();
284	while let Some(node) = queue.pop() {293	while let Some(node) = queue.pop() {
285		if tainted294		if tainted
286			.entry(node.clone())295			.entry(node.clone())
298	}307	}
299}308}
300309
301fn path_to_root(graph: &DrvGraph, from: &str) -> Vec<String> {310fn path_to_root(graph: &DrvGraph, from: &Utf8Path) -> Vec<String> {
302	let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();311	let mut dependents: HashMap<&Utf8Path, Vec<&Utf8Path>> = HashMap::new();
303	for (path, node) in &graph.nodes {312	for (path, node) in &graph.nodes {
304		for dep in node.input_drvs.keys() {313		for dep in node.input_drvs.keys() {
305			dependents314			dependents.entry(dep).or_default().push(path);
306				.entry(dep.as_str())
307				.or_default()
308				.push(path.as_str());
309		}315		}
310	}316	}
311317
312	let mut chain: Vec<String> = vec![node_name(graph, from)];318	let mut chain: Vec<String> = vec![node_name(graph, from)];
313	let mut cur = from;319	let mut cur = from;
314	let mut seen: HashSet<&str> = HashSet::new();320	let mut seen: HashSet<&Utf8Path> = HashSet::new();
315	seen.insert(cur);321	seen.insert(cur);
316	while cur != graph.root.as_str() {322	while cur != graph.root.as_str() {
317		let Some(next) = dependents.get(cur).and_then(|v| v.first().copied()) else {323		let Some(next) = dependents.get(cur).and_then(|v| v.first().copied()) else {
326	chain332	chain
327}333}
328334
329fn node_name(graph: &DrvGraph, path: &str) -> String {335fn node_name(graph: &DrvGraph, path: &Utf8Path) -> String {
330	graph336	graph
331		.nodes337		.nodes
332		.get(path)338		.get(path)
333		.map(|n| n.name.clone())339		.map(|n| n.name.clone())
334		.unwrap_or_else(|| path.to_owned())340		.unwrap_or_else(|| path.to_string())
335}341}
336342
337fn collect_substitute_paths(343fn collect_substitute_paths(
338	graph: &DrvGraph,344	graph: &DrvGraph,
339	wanted: &HashMap<String, Vec<String>>,345	wanted: &HashMap<Utf8PathBuf, Vec<String>>,
340) -> Vec<String> {346) -> Vec<Utf8PathBuf> {
341	let mut paths: HashSet<String> = HashSet::new();347	let mut paths: HashSet<Utf8PathBuf> = HashSet::new();
342	for node in graph.nodes.values() {348	for node in graph.nodes.values() {
343		for src in &node.input_srcs {349		for src in &node.input_srcs {
344			paths.insert(src.clone());350			paths.insert(src.clone());