7 files changed
1#![recursion_limit = "512"]23pub(crate) mod cmds;45pub(crate) mod extra_args;67use std::{process::ExitCode, sync::Arc};89use anyhow::{Context as _, Result, bail};10use camino::Utf8PathBuf;11use clap::{CommandFactory, Parser};12use cmds::{13 build_systems::{BuildSystems, Deploy},14 complete::Complete,15 info::Info,16 rollback::RollbackSingle,17 secrets::Secret,18 tf::Tf,19};20use fleet_base::{host::Config, opts::FleetOpts};21use futures::{TryStreamExt, stream::FuturesUnordered};22#[cfg(feature = "indicatif")]23use human_repr::HumanCount;24#[cfg(feature = "indicatif")]25use indicatif::{ProgressState, ProgressStyle};26use nix_eval::{27 add_file_to_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries,28 init_tokio_for_nix,29};30use opentelemetry::trace::TracerProvider;31use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;32use opentelemetry_exporter_env::{33 OtlpBaseSettings, OtlpLogsSettings, OtlpTracesSettings, ResolvedOtlpSettings,34};35use opentelemetry_sdk::{logs::SdkLoggerProvider, trace::SdkTracerProvider};36use tracing::{Instrument, error, info, info_span};37#[cfg(feature = "indicatif")]38use tracing_indicatif::IndicatifLayer;39use tracing_subscriber::{EnvFilter, prelude::*};4041#[derive(Parser)]42struct Prefetch {}43impl Prefetch {44 async fn run(&self, config: &Config) -> Result<()> {45 let mut prefetch_dir = config.directory.to_path_buf();46 prefetch_dir.push("prefetch");47 if !prefetch_dir.is_dir() {48 info!("nothing to prefetch: no prefetch directory");49 return Ok(());50 }51 let tasks = FuturesUnordered::new();52 for entry in std::fs::read_dir(&prefetch_dir)? {53 let entry = entry?;54 if !entry.metadata()?.is_file() {55 bail!("only files should exist in prefetch directory");56 }57 let name = entry.file_name().to_string_lossy().into_owned();58 let path =59 Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;60 let span = info_span!("prefetching", name = %name);61 tasks.push(async move {62 let added = tokio::task::spawn_blocking(move || add_file_to_store(&name, &path))63 .instrument(span.clone())64 .await??;65 let _g = span.enter();66 info!("{} -> {}", added.hash, added.store_path);67 anyhow::Ok(())68 });69 }70 tasks.try_collect::<Vec<()>>().await?;71 Ok(())72 }73}7475#[derive(Parser)]76enum Opts {77 78 BuildSystems(BuildSystems),79 80 Deploy(Deploy),81 82 RollbackSingle(RollbackSingle),83 84 #[clap(subcommand)]85 Secret(Secret),86 87 Prefetch(Prefetch),88 89 Info(Info),90 91 #[clap(hide(true))]92 Complete(Complete),93 94 Tf(Tf),95}9697#[derive(Parser)]98#[clap(version, author)]99struct RootOpts {100 #[clap(flatten)]101 fleet_opts: FleetOpts,102 #[clap(subcommand)]103 command: Opts,104 #[clap(long, next_help_heading = "Telemetry", env = "OTEL_FLEET")]105 otel: bool,106 #[clap(flatten)]107 otlp_base: OtlpBaseSettings,108 #[clap(flatten)]109 otel_logs: OtlpLogsSettings,110 #[clap(flatten)]111 otel_traces: OtlpTracesSettings,112}113114async fn run_command(config: &Config, opts: FleetOpts, command: Opts) -> Result<()> {115 match command {116 Opts::BuildSystems(c) => c.run(config, &opts).await?,117 Opts::Deploy(d) => d.run(config, &opts).await?,118 Opts::RollbackSingle(r) => r.run(config, &opts).await?,119 Opts::Secret(s) => s.run(config, &opts).await?,120 Opts::Info(i) => i.run(config).await?,121 Opts::Prefetch(p) => p.run(config).await?,122 Opts::Tf(t) => t.run(config).await?,123 124 Opts::Complete(c) => {125 tokio::task::spawn_blocking(move || c.run(RootOpts::command())).await?126 }127 };128 Ok(())129}130131fn setup_logging(opts: &RootOpts) -> Result<()> {132 #[cfg(feature = "indicatif")]133 let indicatif_layer = {134 use std::time::Duration;135136 IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))137 .with_progress_style(138 ProgressStyle::with_template(139 "{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{download_progress} {elapsed}{color_end}",140 )141 .unwrap()142 .with_key("download_progress", |state: &ProgressState, writer: &mut dyn std::fmt::Write| {143 let Some(len) = state.len() else {144 return;145 };146 let pos = state.pos();147 if pos > len {148 let _ = write!(writer, "{}", pos.human_count_bare());149 } else {150 let _ = write!(writer, "{} / {}", pos.human_count_bare(), len.human_count_bare());151 }152 })153 .with_key(154 "color_start",155 |state: &ProgressState, writer: &mut dyn std::fmt::Write| {156 let elapsed = state.elapsed();157158 if elapsed > Duration::from_secs(60) {159 160 let _ = write!(writer, "\x1b[{}m", 1 + 30);161 } else if elapsed > Duration::from_secs(30) {162 163 let _ = write!(writer, "\x1b[{}m", 3 + 30);164 }165 },166 )167 .with_key(168 "color_end",169 |state: &ProgressState, writer: &mut dyn std::fmt::Write| {170 if state.elapsed() > Duration::from_secs(30) {171 let _ = write!(writer, "\x1b[0m");172 }173 },174 ),175 )176 };177178 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));179180 let reg = tracing_subscriber::registry().with({181 let sub = tracing_subscriber::fmt::layer().without_time();182 #[cfg(feature = "indicatif")]183 let sub = sub.with_writer(indicatif_layer.get_stderr_writer());184 sub.with_filter(filter) 185 });186187 #[cfg(feature = "indicatif")]188 let reg = reg.with(indicatif_layer);189190 if opts.otel {191 let traces = ResolvedOtlpSettings::traces(&opts.otlp_base, &opts.otel_traces)?;192 let span_exporter = traces.span_exporter()?;193 let logs = ResolvedOtlpSettings::logs(&opts.otlp_base, &opts.otel_logs)?;194 let log_exporter = logs.log_exporter()?;195196 let span_provider = SdkTracerProvider::builder()197 .with_batch_exporter(span_exporter)198 .build();199 let log_provider = SdkLoggerProvider::builder()200 .with_batch_exporter(log_exporter)201 .build();202203 let logger = OpenTelemetryTracingBridge::new(&log_provider);204 let tracer = span_provider.tracer("fleet");205206 let reg = reg207 .with(tracing_opentelemetry::layer().with_tracer(tracer))208 .with(logger);209210 reg.init();211 } else {212 reg.init();213 };214215 Ok(())216}217218fn main() -> ExitCode {219 let opts = RootOpts::parse();220 if let Opts::Complete(c) = &opts.command {221 c.run(RootOpts::command());222 return ExitCode::SUCCESS;223 }224225 if let Err(e) = setup_logging(&opts) {226 eprintln!("{e:#}");227 return ExitCode::FAILURE;228 }229230 init_libraries();231232 let runtime = tokio::runtime::Builder::new_multi_thread()233 .enable_all()234 .on_thread_start(|| {235 gc_register_my_thread();236 })237 .on_thread_stop(|| {238 gc_unregister_my_thread();239 })240 .build()241 .expect("failed to build runtime");242 let runtime = Arc::new(runtime);243244 init_tokio_for_nix(runtime.clone());245246 runtime.block_on(async {247 tokio::task::spawn(async move {248 if let Err(e) = main_real(opts).await {249 error!("{e:#}");250 ExitCode::FAILURE251 } else {252 ExitCode::SUCCESS253 }254 })255 .await256 .expect("primary task panicked")257 })258 259}260261async fn main_real(opts: RootOpts) -> Result<()> {262 let config = opts.fleet_opts.build(matches!(263 opts.command,264 Opts::Deploy(_) | Opts::BuildSystems(_)265 ))?;266267 match run_command(&config, opts.fleet_opts, opts.command).await {268 Ok(()) => {269 config.save()?;270 Ok(())271 }272 Err(e) => {273 let _ = config.save();274 Err(e)275 }276 }277}278279#[cfg(test)]280mod tests {281 use super::*;282283 #[test]284 fn verify_command() {285 use clap::CommandFactory;286 RootOpts::command().debug_assert();287 }288}
--- 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<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()
}
--- 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) {
--- 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()
--- 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<String>,
},
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<Store>,
parallelism: usize,
events: broadcast::Sender<BuildEvent>,
}
@@ -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<String, Vec<String>>,
+ wanted: &HashMap<Utf8PathBuf, Vec<String>>,
) -> 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<DrvGraph>,
- wanted: HashMap<String, Vec<String>>,
+ wanted: HashMap<Utf8PathBuf, Vec<String>>,
) -> Result<()> {
- let mut indeg: HashMap<String, usize> = graph
+ let mut indeg: HashMap<Utf8PathBuf, usize> = graph
.nodes
.iter()
.map(|(k, n)| (k.clone(), n.input_drvs.len()))
.collect();
- let mut dependents: HashMap<String, Vec<String>> = HashMap::new();
+ let mut dependents: HashMap<Utf8PathBuf, Vec<Utf8PathBuf>> = 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<String> = indeg
+ let mut ready: Vec<Utf8PathBuf> = indeg
.iter()
.filter(|(_, d)| **d == 0)
.map(|(k, _)| k.clone())
.collect();
let mut in_flight = FuturesUnordered::new();
- let mut failed: HashMap<String, String> = HashMap::new();
+ let mut failed: HashMap<Utf8PathBuf, String> = HashMap::new();
// Tainted = transitively depends on a failed drv
- let mut tainted: HashMap<String, String> = HashMap::new();
+ let mut tainted: HashMap<Utf8PathBuf, Utf8PathBuf> = HashMap::new();
loop {
- let batch: Vec<String> = std::mem::take(&mut ready);
+ let batch: Vec<Utf8PathBuf> = 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<String, Vec<String>>,
- indeg: &mut HashMap<String, usize>,
- ready: &mut Vec<String>,
- finished: &str,
+ dependents: &HashMap<Utf8PathBuf, Vec<Utf8PathBuf>>,
+ indeg: &mut HashMap<Utf8PathBuf, usize>,
+ ready: &mut Vec<Utf8PathBuf>,
+ finished: &Utf8Path,
) {
if let Some(deps) = dependents.get(finished) {
for d in deps {
@@ -276,11 +285,11 @@
}
fn mark_tainted(
- dependents: &HashMap<String, Vec<String>>,
- failed: &str,
- tainted: &mut HashMap<String, String>,
+ dependents: &HashMap<Utf8PathBuf, Vec<Utf8PathBuf>>,
+ failed: &Utf8Path,
+ tainted: &mut HashMap<Utf8PathBuf, Utf8PathBuf>,
) {
- let mut queue: Vec<String> = dependents.get(failed).cloned().unwrap_or_default();
+ let mut queue: Vec<Utf8PathBuf> = 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<String> {
- let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();
+fn path_to_root(graph: &DrvGraph, from: &Utf8Path) -> Vec<String> {
+ 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<String> = 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<String, Vec<String>>,
-) -> Vec<String> {
- let mut paths: HashSet<String> = HashSet::new();
+ wanted: &HashMap<Utf8PathBuf, Vec<String>>,
+) -> Vec<Utf8PathBuf> {
+ let mut paths: HashSet<Utf8PathBuf> = 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<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()))
}
}