difftreelog
feat attic
22 files changed
Cargo.lockdiffbeforeafterboth--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5084,6 +5084,7 @@
"remowt-ui-prompt",
"tokio",
"tracing",
+ "tracing-journald",
"tracing-subscriber",
]
@@ -6576,6 +6577,17 @@
]
[[package]]
+name = "tracing-journald"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d3a81ed245bfb62592b1e2bc153e77656d94ee6a0497683a65a12ccaf2438d0"
+dependencies = [
+ "libc",
+ "tracing-core",
+ "tracing-subscriber",
+]
+
+[[package]]
name = "tracing-log"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
Cargo.tomldiffbeforeafterboth--- a/Cargo.toml
+++ b/Cargo.toml
@@ -75,6 +75,7 @@
tokio = { version = "1.45.1", features = ["fs", "macros", "rt", "rt-multi-thread", "sync", "time"] }
tracing = "0.1"
tracing-indicatif = "0.3.13"
+tracing-journald = "0.3.2"
tracing-opentelemetry = "0.33.0"
uuid = { version = "1", features = ["v4"] }
# For fixed coloring needs to be updated to https://github.com/tokio-rs/tracing/pull/3484
cmds/fleet/src/cmds/build_systems.rsdiffbeforeafterboth--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -14,6 +14,8 @@
use tokio::task::spawn_blocking;
use tracing::{Instrument, error, field, info, info_span, warn};
+const TARGET: &str = "build-systems";
+
#[derive(Parser)]
pub struct Deploy {
/// Disable automatic rollback
@@ -32,7 +34,7 @@
}
async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<Utf8PathBuf> {
- info!("building");
+ info!(target: TARGET, "building");
let host = config.host(&hostname)?;
// let action = Action::from(self.subcommand.clone());
let nixos = host.nixos_config()?;
@@ -43,7 +45,7 @@
// We already have system profiles for backups.
if !host.local {
- info!("adding gc root");
+ info!(target: TARGET, "adding gc root");
let local = config.local_host();
let plugin_id = local.ensure_nix_plugin().await?;
let nix = local
@@ -70,7 +72,7 @@
let build_attr = self.build_attr.clone();
for host in hosts {
let config = config.clone();
- let span = info_span!("build", host = field::display(&host.name));
+ let span = info_span!(target: TARGET, "build", host = field::display(&host.name));
let hostname = host.name;
let build_attr = build_attr.clone();
tasks.push(
@@ -78,7 +80,7 @@
let built = match build_task(config, hostname.clone(), &build_attr).await {
Ok(path) => path,
Err(e) => {
- error!("failed to deploy host: {:?}", e);
+ error!(target: TARGET, "failed to deploy host: {:?}", e);
return;
}
};
@@ -86,9 +88,9 @@
let mut out = current_dir().expect("cwd exists");
out.push(format!("built-{hostname}"));
- info!("linking iso image to {:?}", out);
+ info!(target: TARGET, "linking iso image to {:?}", out);
if let Err(e) = symlink(built, out) {
- error!("failed to symlink: {e}")
+ error!(target: TARGET, "failed to symlink: {e}")
}
})
.instrument(span),
@@ -105,7 +107,7 @@
let tasks = FuturesUnordered::new();
for host in hosts.into_iter() {
let config = config.clone();
- let span = info_span!("deploy", host = field::display(&host.name));
+ let span = info_span!(target: TARGET, "deploy", host = field::display(&host.name));
let hostname = host.name.clone();
let opts = opts.clone();
if let Some(deploy_kind) = opts.action_attr::<DeployKind>(&host, "deploy_kind")? {
@@ -125,7 +127,7 @@
{
Ok(path) => path,
Err(e) => {
- error!("failed to build host system closure: {:?}", e);
+ error!(target: TARGET, "failed to build host system closure: {:?}", e);
return;
}
};
@@ -133,7 +135,7 @@
let deploy_kind = match host.deploy_kind().await {
Ok(v) => v,
Err(e) => {
- error!("failed to query target deploy kind: {e}");
+ error!(target: TARGET, "failed to query target deploy kind: {e}");
return;
}
};
@@ -141,7 +143,7 @@
// TODO: Make disable_rollback a host attribute instead
let mut disable_rollback = self.disable_rollback;
if !disable_rollback && deploy_kind != DeployKind::Fleet {
- warn!("disabling rollback, as not supported by non-fleet deployment kinds");
+ warn!(target: TARGET, "disabling rollback, as not supported by non-fleet deployment kinds");
disable_rollback = true;
}
@@ -149,7 +151,7 @@
match upload_task(&host, GenerationStorage::Deployer, built).await {
Ok(v) => v,
Err(e) => {
- error!("upload failed: {e}");
+ error!(target: TARGET, "upload failed: {e}");
return;
}
};
@@ -169,7 +171,7 @@
)
.await
{
- error!("activation failed: {e}");
+ error!(target: TARGET, "activation failed: {e}");
}
})
.instrument(span),
cmds/fleet/src/log_tree.rsdiffbeforeafterboth--- /dev/null
+++ b/cmds/fleet/src/log_tree.rs
@@ -0,0 +1,243 @@
+//! Streaming, indented tree formatter.
+//!
+//! Unlike `tracing-tree` it keeps no per-thread depth state, so interleaved
+//! parallel output stays correct; unlike `tracing-forest` it doesn't buffer
+//! whole trees, so logs stream live. Visual style mirrors `@meteor-it/logger`'s
+//! node receiver: a fixed-width, right-aligned, level-colored target column,
+//! then a magenta `>`/`<` gutter for span enter/exit (with duration on exit),
+//! and 2-space indentation per depth for events.
+
+use std::{
+ fmt::Write as _,
+ io::{IsTerminal as _, Write as _},
+ sync::{
+ Mutex,
+ atomic::{AtomicBool, Ordering},
+ },
+ time::{Duration, Instant},
+};
+
+use tracing::{Event, Level, Subscriber, field::Field, field::Visit, span::Attributes, span::Id};
+use tracing_subscriber::{
+ fmt::MakeWriter,
+ layer::{Context, Layer},
+ registry::{LookupSpan, Scope},
+};
+
+/// Width of the right-aligned target column.
+const NAME_LIMIT: usize = 18;
+
+#[derive(Default)]
+struct Fields {
+ message: String,
+ rest: String,
+}
+
+impl Visit for Fields {
+ fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+ if field.name() == "message" {
+ let _ = write!(self.message, "{value:?}");
+ } else {
+ let _ = write!(self.rest, " {}={:?}", field.name(), value);
+ }
+ }
+}
+
+impl Fields {
+ fn render(&self) -> String {
+ format!("{}{}", self.message, self.rest).trim().to_owned()
+ }
+}
+
+/// Per-span state stashed in the registry's extensions.
+struct SpanData {
+ fields: String,
+ started: Instant,
+ // Whether the span's enter line was written, so we only emit an exit line
+ // for spans that actually appeared.
+ opened: AtomicBool,
+}
+
+/// SGR code → escape for the level-colored target column.
+fn level_code(level: &Level) -> &'static str {
+ match *level {
+ Level::ERROR => "31m", // red
+ Level::WARN => "33m", // yellow
+ Level::INFO => "34m", // blue
+ Level::DEBUG => "39m", // default fg
+ Level::TRACE => "90m", // gray
+ }
+}
+
+/// 2-space indent per depth, then a gutter symbol.
+fn gutter(depth: usize, sym: char) -> String {
+ let mut s = " ".repeat(depth);
+ s.push(sym);
+ s
+}
+
+fn fmt_dur(d: Duration) -> String {
+ if d.as_secs_f64() >= 1.0 {
+ format!("{:.2}s", d.as_secs_f64())
+ } else if d.as_millis() >= 1 {
+ format!("{}ms", d.as_millis())
+ } else {
+ format!("{}µs", d.as_micros())
+ }
+}
+
+pub struct TreeLayer<W> {
+ writer: W,
+ color: bool,
+ // Serializes the whole compute-and-write so enter/exit decisions, output
+ // ordering, and the writes stay consistent under parallel emission.
+ last: Mutex<Vec<Id>>,
+}
+
+impl<W> TreeLayer<W> {
+ pub fn new(writer: W) -> Self {
+ Self {
+ writer,
+ color: std::io::stderr().is_terminal() && std::env::var_os("NO_COLOR").is_none(),
+ last: Mutex::new(vec![]),
+ }
+ }
+
+ /// Raw SGR escape, or empty when color is disabled.
+ fn sgr(&self, code: &str) -> String {
+ if self.color {
+ format!("\x1b[{code}")
+ } else {
+ String::new()
+ }
+ }
+
+ fn name_col(&self, mut target: &str, code: &str) -> String {
+ let w = NAME_LIMIT;
+ if target.len() > w {
+ target = &target[..w];
+ }
+ if self.color {
+ format!("\x1b[{code}\x1b[1m{target:>w$}\x1b[0m")
+ } else {
+ format!("{target:>w$}")
+ }
+ }
+
+ fn print_ancestors<S>(&self, scope: Option<Scope<'_, S>>, out: &mut String) -> usize
+ where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+ {
+ let scope: Vec<_> = scope.map(|s| s.from_root().collect()).unwrap_or_default();
+
+ let mut _last = self.last.lock().expect("not poisoned");
+
+ let skip = scope
+ .iter()
+ .enumerate()
+ .take_while(|(i, s)| _last.get(*i) == Some(&s.id()))
+ .count();
+
+ for (depth, span) in scope.iter().enumerate().skip(skip) {
+ let ext = span.extensions();
+ let Some(data) = ext.get::<SpanData>() else {
+ continue;
+ };
+ let newly_open = !data.opened.swap(true, Ordering::Relaxed);
+ let _ = writeln!(
+ out,
+ "{} {}{}{} {}{}{}",
+ self.name_col(span.metadata().target(), "44m"),
+ self.sgr("35m"),
+ gutter(depth, if newly_open { '>' } else { '.' }),
+ self.sgr("1m"),
+ span.name(),
+ data.fields,
+ self.sgr("0m"),
+ );
+ }
+
+ *_last = scope.iter().map(|v| v.id()).collect();
+
+ scope.len()
+ }
+}
+
+impl<S, W> Layer<S> for TreeLayer<W>
+where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+ W: for<'a> MakeWriter<'a> + 'static,
+{
+ fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
+ let span = ctx.span(id).expect("new span exists");
+ let mut fields = Fields::default();
+ attrs.record(&mut fields);
+ span.extensions_mut().insert(SpanData {
+ fields: fields.rest,
+ started: Instant::now(),
+ opened: AtomicBool::new(false),
+ });
+ }
+
+ fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
+ let mut out = String::new();
+
+ let scope = ctx.event_scope(event);
+ let pad = self.print_ancestors(scope, &mut out);
+
+ let mut fields = Fields::default();
+ event.record(&mut fields);
+ let message = fields.render();
+ let meta = event.metadata();
+ for (i, message) in message.split('\n').enumerate() {
+ // Event line: black bg + colored target column, then indent + message.
+ let _ = writeln!(
+ out,
+ "{}{}{}{}{}",
+ self.sgr("40m"),
+ self.name_col(
+ if i == 0 { meta.target() } else { "|" },
+ level_code(meta.level())
+ ),
+ self.sgr("0m"),
+ gutter(pad, ' '),
+ message,
+ );
+ }
+
+ let _ = self.writer.make_writer().write_all(out.as_bytes());
+ }
+
+ fn on_close(&self, id: Id, ctx: Context<'_, S>) {
+ let Some(span) = ctx.span(&id) else { return };
+ let (opened, dur) = span
+ .extensions()
+ .get::<SpanData>()
+ .map_or((false, Duration::ZERO), |d| {
+ (d.opened.load(Ordering::Relaxed), d.started.elapsed())
+ });
+ if !opened {
+ return;
+ }
+
+ let mut out = String::new();
+ let _pad = self.print_ancestors(ctx.span_scope(&id), &mut out);
+
+ let depth = span.scope().skip(1).count(); // ancestors == depth
+ let _ = writeln!(
+ out,
+ "{} {}{}{} {}{} (Done in {}){}",
+ self.name_col(span.metadata().target(), "44m"),
+ self.sgr("35m"),
+ gutter(depth, '<'),
+ self.sgr("1m"),
+ span.name(),
+ self.sgr("22m"),
+ fmt_dur(dur),
+ self.sgr("0m"),
+ );
+
+ let _guard = self.last.lock().expect("not poisoned");
+ let _ = self.writer.make_writer().write_all(out.as_bytes());
+ }
+}
cmds/fleet/src/main.rsdiffbeforeafterboth--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -1,6 +1,7 @@
#![recursion_limit = "512"]
pub(crate) mod cmds;
+mod log_tree;
use std::{process::ExitCode, sync::Arc};
@@ -21,6 +22,7 @@
use human_repr::HumanCount;
#[cfg(feature = "indicatif")]
use indicatif::{ProgressState, ProgressStyle};
+use log_tree::TreeLayer;
use nix_eval::{
eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
};
@@ -128,15 +130,21 @@
fn setup_logging(opts: &RootOpts) -> Result<()> {
#[cfg(feature = "indicatif")]
let indicatif_layer = {
+ use std::fmt;
use std::time::Duration;
IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))
+ .with_span_child_prefix_indent(" ")
+ .with_span_child_prefix_symbol("")
.with_progress_style(
ProgressStyle::with_template(
- "{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{download_progress} {elapsed}{color_end}",
+ "{span_child_prefix:.magenta}{chevron:.magenta}{span_name:<18.blue.bold} {wide_msg} {span_fields:.dim} {color_start}{download_progress} {elapsed}{color_end}",
)
.unwrap()
- .with_key("download_progress", |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ .with_key("chevron", |_: &ProgressState, writer: &mut dyn fmt::Write| {
+ let _ = write!(writer, "> ");
+ })
+ .with_key("download_progress", |state: &ProgressState, writer: &mut dyn fmt::Write| {
let Some(len) = state.len() else {
return;
};
@@ -149,7 +157,7 @@
})
.with_key(
"color_start",
- |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ |state: &ProgressState, writer: &mut dyn fmt::Write| {
let elapsed = state.elapsed();
if elapsed > Duration::from_secs(60) {
@@ -163,7 +171,7 @@
)
.with_key(
"color_end",
- |state: &ProgressState, writer: &mut dyn std::fmt::Write| {
+ |state: &ProgressState, writer: &mut dyn fmt::Write| {
if state.elapsed() > Duration::from_secs(30) {
let _ = write!(writer, "\x1b[0m");
}
@@ -174,12 +182,15 @@
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
- let reg = tracing_subscriber::registry().with({
- let sub = tracing_subscriber::fmt::layer().without_time();
+ let tree = {
#[cfg(feature = "indicatif")]
- let sub = sub.with_writer(indicatif_layer.get_stderr_writer());
- sub.with_filter(filter) // .without,
- });
+ let writer = indicatif_layer.get_stderr_writer();
+ #[cfg(not(feature = "indicatif"))]
+ let writer = || std::io::stderr();
+ TreeLayer::new(writer)
+ };
+
+ let reg = tracing_subscriber::registry().with(tree.with_filter(filter));
#[cfg(feature = "indicatif")]
let reg = reg.with(indicatif_layer);
@@ -200,14 +211,12 @@
let logger = OpenTelemetryTracingBridge::new(&log_provider);
let tracer = span_provider.tracer("fleet");
- let reg = reg
- .with(tracing_opentelemetry::layer().with_tracer(tracer))
- .with(logger);
-
- reg.init();
+ reg.with(tracing_opentelemetry::layer().with_tracer(tracer))
+ .with(logger)
+ .init();
} else {
reg.init();
- };
+ }
Ok(())
}
@@ -252,7 +261,6 @@
.await
.expect("primary task panicked")
})
- // async_main(opts)
}
async fn main_real(opts: RootOpts) -> Result<()> {
crates/fleet-base/src/deploy.rsdiffbeforeafterboth--- a/crates/fleet-base/src/deploy.rs
+++ b/crates/fleet-base/src/deploy.rs
@@ -290,8 +290,10 @@
if !host.local {
info!("uploading system closure");
let mut tries = 0;
+ // TODO: Use gc prefix?..
loop {
- match host.remote_derivation(&generation).await {
+ let name = format!("{}-system", host.gc_root_prefix());
+ match host.remote_derivation(&generation, Some(&name)).await {
Ok(remote) => {
assert!(remote == generation, "CA derivations aren't implemented");
return Ok(remote);
crates/fleet-base/src/host.rsdiffbeforeafterboth--- a/crates/fleet-base/src/host.rs
+++ b/crates/fleet-base/src/host.rs
@@ -13,16 +13,16 @@
use camino::{Utf8Path, Utf8PathBuf};
use chrono::{DateTime, Utc};
use fleet_shared::SecretData;
-use nix_eval::{Store, Value, drv::DrvGraph, eval_store, 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_fleet::NixClient;
-use remowt_link_shared::BifConfig;
+use remowt_link_shared::{Address, BifConfig};
use remowt_ui_prompt::auto::AutoPrompter;
use remowt_ui_prompt::bifrost::PromptEndpoints;
use remowt_ui_prompt::{PrependSourcePrompter, Source};
use tempfile::NamedTempFile;
-use tokio::{sync::OnceCell, task::spawn_blocking};
+use tokio::{process::Command, sync::OnceCell, task::spawn_blocking};
use tracing::{info, instrument, warn};
use crate::fleetdata::{
@@ -104,11 +104,14 @@
session_destination: OnceLock<String>,
legacy_ssh_store: OnceLock<bool>,
+ /// fleetConfiguration.hosts.host, None for local (as local is only used for local tool overrides)
pub host_config: Option<Value>,
pub nixos_config: OnceLock<Value>,
pub nixos_unchecked_config: OnceLock<Value>,
pub pkgs_override: Option<Value>,
+ binary_cache: OnceCell<Option<BinaryCache>>,
+
// TODO: Move command helpers away with connectivity refactor
pub local: bool,
pub remowt: OnceCell<Remowt>,
@@ -131,13 +134,35 @@
AgentBundle::from_dir(agents_dir()?)
}
-enum RemoteDerivationMode {
- /// Closure is (pre)signed, remote host is expected to trust our signatures.
- CopySigned,
- /// Closure is not signed, using escalated nix daemon to bypass signature verification.
- CopyPrivileged,
- /// Closure is uploaded to attic, remote host is expected to trust its signature.
- Attic { name: String },
+#[derive(Clone)]
+pub enum BinaryCache {
+ Attic { name: String, pin: bool },
+}
+impl BinaryCache {
+ async fn upload(&self, local: &Remowt, path: &Utf8Path, pin_name: Option<&str>) -> Result<()> {
+ match self {
+ BinaryCache::Attic { name, pin } => {
+ let mut cmd = local.cmd("attic");
+ cmd.arg("push").arg(name).arg(path.as_str());
+ cmd.run()
+ .await
+ .context("failed to push path to attic (is attic-cli on PATH?)")?;
+
+ if *pin && let Some(pin_name) = pin_name {
+ let mut cmd = local.cmd("attic");
+ cmd.arg("pin")
+ .arg("create")
+ .arg(name)
+ .arg(pin_name)
+ .arg(path.as_str());
+ cmd.run()
+ .await
+ .context("failed to pin path in attic (is your attic-cli compatible?)")?;
+ }
+ Ok(())
+ }
+ }
+ }
}
impl ConfigHost {
@@ -156,6 +181,32 @@
.set(legacy)
.expect("legacy ssh store is already set")
}
+ pub fn binary_cache(&self) -> Result<Option<BinaryCache>> {
+ if let Some(v) = self.binary_cache.get() {
+ return Ok(v.clone());
+ }
+ let cache = if let Some(host_config) = &self.host_config {
+ let cache = nix_go!(host_config.cache);
+ let enable: bool = nix_go_json!(cache.enable);
+ if enable {
+ let kind: String = nix_go_json!(cache.kind);
+ let name: String = nix_go_json!(cache.name);
+ match kind.as_str() {
+ "attic" => {
+ let pin: bool = nix_go_json!(cache.pin);
+ Some(BinaryCache::Attic { name, pin })
+ }
+ v => bail!("unknown binary cache kind: {v}"),
+ }
+ } else {
+ None
+ }
+ } else {
+ None
+ };
+ let _ = self.binary_cache.set(cache.clone());
+ Ok(cache)
+ }
pub async fn deploy_kind(&self) -> Result<DeployKind> {
self.deploy_kind.get_or_try_init(|| async {
let remowt = self.remowt().await?;
@@ -238,8 +289,10 @@
let built = plugin
.build("out")
.context("failed to build the fleet nix plugin")?;
+ let pin = format!("{}-remowt-plugin-fleet", self.gc_root_prefix());
let copied = self
- .remote_derivation(&built)
+ // TODO: Pin? But this pin will be host-specific, and heavy because of nix dependency...
+ .remote_derivation(&built, Some(&pin))
.await
.context("failed to copy the fleet nix plugin to the host store")?;
let bin = copied.join("bin/remowt-plugin-fleet");
@@ -248,6 +301,13 @@
.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())
.await
.context("failed to load the fleet nix plugin")?;
+ // Should not fail, as plugin loading waits for connection on remote, and we only wait for forward... On the other hand, we should monitor connection to the remote itself somewhere too.
+ self.remowt()
+ .await?
+ .rpc()
+ .wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))
+ .await
+ .map_err(|_| anyhow!("failed to wait"))?;
anyhow::Ok(())
})
.await?;
@@ -339,17 +399,31 @@
}
/// Returns path for futureproofing, as path might change i.e on conversion to CA
#[instrument(skip(self))]
- pub async fn remote_derivation(&self, path: &Utf8Path) -> Result<Utf8PathBuf> {
+ pub async fn remote_derivation(
+ &self,
+ path: &Utf8Path,
+ pin_name: Option<&str>,
+ ) -> Result<Utf8PathBuf> {
let path = path.to_owned();
if self.local {
// Path is located locally, thus already trusted.
return Ok(path);
}
+ let cache = self.binary_cache()?;
+ let mut cache = if let Some(cache) = cache {
+ let local = self.config.local_host().remowt().await?;
+ let path = path.to_owned();
+ let pin_name = pin_name.map(str::to_owned);
+ Some(tokio::spawn(async move {
+ if let Err(e) = cache.upload(&local, &path, pin_name.as_deref()).await {
+ warn!("failed to upload closure {path} to cache: {e:?}");
+ }
+ }))
+ } else {
+ None
+ };
let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {
let path = path.clone();
- let graph = DrvGraph::resolve(&eval_store(), &path)
- .context("failed to resolve graph to be uploaded")?;
- info!("signing {} paths", graph.nodes.len());
Box::pin(async move {
let local = self.config.local_host();
@@ -377,13 +451,62 @@
.expect("copy_to panicked")
.context("copying closure to remote store")?;
}
+
+ if let Some(cache) = cache.take() {
+ let _ = cache.await;
+ }
+
Ok(path)
}
+
+ /// Like [`Self::remote_derivation`], but instead of copying the closure over
+ /// the nix daemon tunnel, uploads it to an attic binary cache and has the
+ /// remote host substitute it from there. The remote is expected to already
+ /// trust the cache's signing key and have it configured as a substituter.
+ #[instrument(skip(self))]
+ pub async fn remote_derivation_attic(
+ &self,
+ path: &Utf8Path,
+ cache_name: &str,
+ ) -> Result<Utf8PathBuf> {
+ let path = path.to_owned();
+ if self.local {
+ // Path is located locally, thus already trusted.
+ return Ok(path);
+ }
+
+ info!("uploading {path} closure to attic cache {cache_name}");
+ let status = Command::new("attic")
+ .arg("push")
+ .arg(cache_name)
+ .arg(path.as_str())
+ .status()
+ .await
+ .context("failed to spawn attic-cli, is it on PATH?")?;
+ ensure!(status.success(), "attic push failed: {status}");
+
+ info!("substituting {path} on the remote host");
+ let nix = self.nix_client().await?;
+ let substituted = nix
+ .substitute(vec![path.clone()])
+ .await
+ .map_err(|e| anyhow!("{e:?}"))?
+ .map_err(|e| anyhow!("{e}"))?;
+ ensure!(
+ substituted.contains(&path),
+ "remote host failed to substitute {path} from attic cache {cache_name}"
+ );
+ Ok(path)
+ }
}
struct HostSecretDefinition(Value);
impl ConfigHost {
+ pub fn gc_root_prefix(&self) -> String {
+ format!("{}-{}", self.config.gc_root_prefix(), self.name)
+ }
+
// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,
// assuming getting tags always returns the same value.
pub fn tags(&self) -> Result<Vec<String>> {
@@ -504,6 +627,7 @@
cell
},
pkgs_override: Some(self.default_pkgs.clone()),
+ binary_cache: OnceCell::new(),
local: true,
remowt: OnceCell::new(),
@@ -546,6 +670,7 @@
nixos_unchecked_config: OnceLock::new(),
groups: OnceLock::new(),
pkgs_override: None,
+ binary_cache: OnceCell::new(),
// TODO: Remove with connectivit refactor
local: self.localhost == name,
crates/fleet-base/src/keys.rsdiffbeforeafterboth--- a/crates/fleet-base/src/keys.rs
+++ b/crates/fleet-base/src/keys.rs
@@ -9,6 +9,9 @@
use crate::{fleetdata::SecretOwner, host::Config};
impl Config {
+ pub fn gc_root_prefix(&self) -> String {
+ self.data.gc_root_prefix.clone()
+ }
fn cached_host_key(&self, host: &str) -> Option<String> {
let hosts = self.data.hosts.read().expect("no poisoning");
let key = hosts.get(host).map(|h| &h.encryption_key);
crates/fleet-base/src/primops.rsdiffbeforeafterboth--- a/crates/fleet-base/src/primops.rs
+++ b/crates/fleet-base/src/primops.rs
@@ -119,7 +119,7 @@
.context("failed to build generator for target host")?;
let generator = host_on
- .remote_derivation(&generator)
+ .remote_derivation(&generator, None)
.await
.context("failed to copy generator to target host")?;
crates/nix-eval/src/lib.rsdiffbeforeafterboth1use std::borrow::Cow;2use std::cell::RefCell;3use std::collections::HashMap;4use std::ffi::{CStr, CString, c_char, c_int, c_uint, c_void};5use std::ptr::{null, null_mut};6use std::sync::{Arc, LazyLock, OnceLock};7use std::{array, fmt, slice};89use anyhow::{Context, anyhow, bail};10use camino::{Utf8Path, Utf8PathBuf};11use itertools::Itertools;12use serde::Serialize;13use serde::de::DeserializeOwned;14use std::mem::transmute;1516pub use anyhow::Result;17use tracing::{Span, instrument, warn};1819use self::logging::{ErrorInfoBuilder, nix_logging_cxx};20use self::nix_cxx::set_fetcher_setting;21use self::nix_raw::{22 BindingsBuilder as c_bindings_builder, EvalState as c_eval_state, GC_SUCCESS,23 GC_allow_register_threads, GC_get_stack_base, GC_register_my_thread, GC_stack_base,24 GC_thread_is_registered, GC_unregister_my_thread, ListBuilder as c_list_builder, PrimOp,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,27 clear_err, copy_value, err_NIX_ERR_KEY, err_NIX_ERR_NIX_ERROR, err_NIX_ERR_OVERFLOW,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,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,32 flake_reference_and_fragment_from_string, flake_reference_parse_flags,33 flake_reference_parse_flags_free, flake_reference_parse_flags_new,34 flake_reference_parse_flags_set_base_directory, flake_settings, flake_settings_free,35 flake_settings_new, gc_now as gc_now_raw, get_attr_byname, get_attr_name_byidx, get_attrs_size,36 get_list_byidx, get_list_size, get_string, get_type, has_attr_byname, init_bool, init_int,37 init_primop, init_string, libexpr_init, libstore_init, libutil_init, list_builder_free,38 list_builder_insert, locked_flake, locked_flake_free, locked_flake_get_output_attrs,39 make_attrs, make_bindings_builder, make_list, make_list_builder, realised_string,40 realised_string_free, realised_string_get_buffer_size, realised_string_get_buffer_start,41 realised_string_get_store_path, realised_string_get_store_path_count, register_primop,42 set_err_msg, setting_set, state_free, store_copy_closure, store_free, store_open,43 store_parse_path, store_path_free, store_path_name, string_realise, value, value_call,44 value_decref, value_force, value_incref,45};4647// Contains macros helpers48pub mod drv;49pub mod logging;50#[doc(hidden)]51pub mod macros;52pub mod scheduler;5354#[doc(hidden)]55pub mod __macro_support {56 pub use std::collections::hash_map::HashMap;5758 pub use anyhow::Context;59 pub use tokio::task::block_in_place;60}61pub mod util;6263#[allow(64 non_upper_case_globals,65 non_camel_case_types,66 non_snake_case,67 dead_code68)]69mod nix_raw {70 include!(concat!(env!("OUT_DIR"), "/bindings.rs"));71}72#[cxx::bridge]73pub mod nix_cxx {74 struct AddFileToStoreResult {75 error: String,76 store_path: String,77 hash: String,78 }79 struct CxxProfileGeneration {80 id: u64,81 store_path: String,82 creation_time_unix: i64,83 current: bool,84 }85 struct CxxListGenerationsResult {86 error: String,87 generations: Vec<CxxProfileGeneration>,88 }89 struct CxxBuildResult {90 error: String,91 outputs: Vec<String>,92 }93 unsafe extern "C++" {94 type nix_fetchers_settings;95 type Store;96 include!("nix-eval/src/lib.hh");9798 #[allow(clippy::missing_safety_doc)]99 unsafe fn set_fetcher_setting(100 settings: *mut nix_fetchers_settings,101 setting: *const c_char,102 value: *const c_char,103 );104105 #[allow(clippy::missing_safety_doc)]106 unsafe fn switch_profile(store: *mut Store, profile: &str, store_path: &str) -> String;107108 #[allow(clippy::missing_safety_doc)]109 unsafe fn sign_closure(store: *mut Store, store_path: &str, key_file: &str) -> String;110111 #[allow(clippy::missing_safety_doc)]112 unsafe fn add_file_to_store(113 store: *mut Store,114 name: &str,115 path: &str,116 ) -> AddFileToStoreResult;117118 fn list_generations(profile_path: &str) -> CxxListGenerationsResult;119120 #[allow(clippy::missing_safety_doc)]121 unsafe fn build_drv_outputs(122 store: *mut Store,123 drv_path: &str,124 output_names_joined: &str,125 ) -> CxxBuildResult;126127 #[allow(clippy::missing_safety_doc)]128 unsafe fn substitute_paths(store: *mut Store, paths_joined: &str) -> CxxBuildResult;129130 #[allow(clippy::missing_safety_doc)]131 unsafe fn is_valid_path(store: *mut Store, path: &str) -> bool;132 }133}134135#[derive(Debug, PartialEq, Eq)]136pub enum NixType {137 Thunk,138 Int,139 Float,140 Bool,141 String,142 Path,143 Null,144 Attrs,145 List,146 Function,147 External,148}149impl NixType {150 fn from_int(c: c_uint) -> Self {151 match c {152 0 => Self::Thunk,153 1 => Self::Int,154 2 => Self::Float,155 3 => Self::Bool,156 4 => Self::String,157 5 => Self::Path,158 6 => Self::Null,159 7 => Self::Attrs,160 8 => Self::List,161 9 => Self::Function,162 10 => Self::External,163 _ => unreachable!("unknown nix type: {c}"),164 }165 }166}167168enum FunctorKind {169 Function,170 Functor,171}172173#[derive(Debug)]174#[repr(i32)]175pub enum NixErrorKind {176 Unknown = err_NIX_ERR_UNKNOWN,177 Overflow = err_NIX_ERR_OVERFLOW,178 Key = err_NIX_ERR_KEY,179 Generic = err_NIX_ERR_NIX_ERROR,180}181impl NixErrorKind {182 fn from_int(v: c_int) -> Option<Self> {183 Some(match v {184 0 => return None,185 nix_raw::err_NIX_ERR_UNKNOWN => Self::Unknown,186 nix_raw::err_NIX_ERR_OVERFLOW => Self::Overflow,187 nix_raw::err_NIX_ERR_KEY => Self::Key,188 nix_raw::err_NIX_ERR_NIX_ERROR => Self::Generic,189 _ => {190 debug_assert!(false, "unexpected nix error kind: {v}");191 Self::Unknown192 }193 })194 }195}196197pub fn gc_now() {198 unsafe { gc_now_raw() };199}200201pub fn gc_register_my_thread() {202 assert_eq!(unsafe { GC_thread_is_registered() }, 0);203204 let mut sb = GC_stack_base {205 mem_base: null_mut(),206 };207 let r = unsafe { GC_get_stack_base(&mut sb) };208 if r as u32 != GC_SUCCESS {209 panic!("failed to get thread stack base");210 }211 unsafe { GC_register_my_thread(&sb) };212}213pub fn gc_unregister_my_thread() {214 assert_eq!(unsafe { GC_thread_is_registered() }, 1);215216 unsafe { GC_unregister_my_thread() };217}218219pub struct ThreadRegisterGuard {}220impl ThreadRegisterGuard {221 #[allow(clippy::new_without_default)]222 pub fn new() -> Self {223 gc_register_my_thread();224 Self {}225 }226}227impl Drop for ThreadRegisterGuard {228 fn drop(&mut self) {229 gc_unregister_my_thread();230 }231}232233#[repr(transparent)]234pub struct NixContext(*mut c_context);235impl NixContext {236 pub fn set_err_raw(&mut self, err: NixErrorKind, msg: &CStr) {237 unsafe { set_err_msg(self.0, err as c_int, msg.as_ptr()) };238 }239 pub fn set_err(&mut self, err: anyhow::Error) {240 let fmt = format!("{err:?}").replace("\0", "\\0");241 self.set_err_raw(242 NixErrorKind::Generic,243 &CString::new(fmt).expect("NUL bytes were just replaced"),244 );245 }246 pub fn new() -> Self {247 let ctx = unsafe { c_context_create() };248 Self(ctx)249 }250 fn error_kind(&self) -> Option<NixErrorKind> {251 let code = unsafe { err_code(self.0) };252 NixErrorKind::from_int(code)253 }254 fn error<'t>(&self) -> Option<(Cow<'t, str>, Option<Box<ErrorInfoBuilder>>)> {255 if let NixErrorKind::Generic = self.error_kind()? {256 let ei = unsafe { logging::nix_logging_cxx::extract_error_info(self.0) };257 let mut err_out = String::new();258 unsafe {259 err_info_msg(260 null_mut(),261 self.0,262 Some(copy_nix_str),263 (&raw mut err_out).cast(),264 )265 };266 return Some((Cow::Owned(err_out), Some(ei)));267 };268269 // TODO: Can throw error (resulting in panic) if unable to retrieve error. Should be able to resolve by passing context as a first argument,270 // but it looks ugly271 let str = unsafe { err_msg(null_mut(), self.0, null_mut()) };272 Some((unsafe { CStr::from_ptr(str) }.to_string_lossy(), None))273 }274 fn clean_err(&mut self) {275 unsafe {276 clear_err(self.0);277 }278 }279280 fn bail_if_error(&self) -> Result<()> {281 if let Some((err, stack)) = self.error() {282 let mut e = Err(anyhow!("{err}"));283 if let Some(stack) = stack {284 for ele in stack.stack_frames {285 e = e.with_context(|| {286 if ele.pos.is_empty() {287 ele.msg288 } else {289 format!("{} at {}", ele.msg, ele.pos)290 }291 })292 }293 }294 return e.context("<nix frames>");295 };296 Ok(())297 }298299 fn run_in_context<T>(&mut self, f: impl FnOnce(*mut c_context) -> T) -> Result<T> {300 self.clean_err();301 let o = f(self.0);302 self.bail_if_error()?;303 self.clean_err();304 Ok(o)305 }306}307308impl Default for NixContext {309 fn default() -> Self {310 Self::new()311 }312}313impl Drop for NixContext {314 fn drop(&mut self) {315 unsafe {316 c_context_free(self.0);317 }318 }319}320struct GlobalState {321 // Store should be valid as long as EvalState is valid322 #[allow(dead_code)]323 store: Arc<Store>,324 state: EvalState,325}326impl GlobalState {327 fn new() -> Result<Self> {328 let mut ctx = NixContext::new();329 let store = Arc::new(330 ctx.run_in_context(|c| unsafe { store_open(c, c"auto".as_ptr(), null_mut()) })331 .map(Store)?,332 );333334 let builder = ctx.run_in_context(|c| unsafe { eval_state_builder_new(c, store.0) })?;335 ctx.run_in_context(|c| unsafe { eval_state_builder_load(c, builder) })?;336 ctx.run_in_context(|c| unsafe {337 eval_state_builder_set_eval_setting(338 c,339 builder,340 c"lazy-trees".as_ptr(),341 c"true".as_ptr(),342 )343 })?;344 ctx.run_in_context(|c| unsafe {345 eval_state_builder_set_eval_setting(346 c,347 builder,348 c"lazy-locks".as_ptr(),349 c"true".as_ptr(),350 )351 })?;352 let state = ctx353 .run_in_context(|c| unsafe { eval_state_build(c, builder) })354 .map(EvalState)?;355356 Ok(Self { store, state })357 }358}359360struct ThreadState {361 ctx: NixContext,362}363impl ThreadState {364 fn new() -> Result<Self> {365 let ctx = NixContext::new();366367 Ok(Self { ctx })368 }369}370371static GLOBAL_STATE: LazyLock<GlobalState> =372 LazyLock::new(|| GlobalState::new().expect("global state init shouldn't fail"));373374thread_local! {375 static THREAD_STATE: RefCell<ThreadState> = RefCell::new(ThreadState::new().expect("thread state init shouldn't fail"));376}377pub(crate) fn with_default_context<T>(378 f: impl FnOnce(*mut c_context, *mut c_eval_state) -> T,379) -> Result<T> {380 let global = &GLOBAL_STATE.state;381 let (ctx, state) = THREAD_STATE.with_borrow_mut(|w| (w.ctx.0, global.0));382 let mut ctx = NixContext(ctx);383 let v = ctx.run_in_context(|c| f(c, state));384 // It is reused for thread385 std::mem::forget(ctx);386 v387}388389pub fn set_setting(s: &CStr, v: &CStr) -> Result<()> {390 with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())391}392393#[derive(Debug)]394pub struct AddedFile {395 pub store_path: Utf8PathBuf,396 pub hash: String,397}398399#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]400pub struct ProfileGeneration {401 pub id: u64,402 pub store_path: Utf8PathBuf,403 pub creation_time_unix: i64,404 pub current: bool,405}406407#[instrument]408pub fn list_generations(profile_path: &str) -> Result<Vec<ProfileGeneration>> {409 let res = nix_cxx::list_generations(profile_path);410 if !res.error.is_empty() {411 bail!(412 "failed to list generations at {profile_path}: {}",413 res.error414 );415 }416 Ok(res417 .generations418 .into_iter()419 .map(|g| ProfileGeneration {420 id: g.id,421 store_path: Utf8PathBuf::from(g.store_path),422 creation_time_unix: g.creation_time_unix,423 current: g.current,424 })425 .collect())426}427428pub struct FetchSettings(*mut fetchers_settings);429impl FetchSettings {430 pub fn new() -> Self {431 Self::try_new().expect("allocation should not fail")432 }433 fn try_new() -> Result<Self> {434 with_default_context(|c, _| unsafe { fetchers_settings_new(c) }).map(Self)435 }436 pub fn set(&mut self, setting: &CStr, value: &CStr) {437 unsafe {438 set_fetcher_setting(self.0.cast(), setting.as_ptr(), value.as_ptr());439 };440 }441}442unsafe impl Send for FetchSettings {}443unsafe impl Sync for FetchSettings {}444445impl Default for FetchSettings {446 fn default() -> Self {447 Self::new()448 }449}450451impl Drop for FetchSettings {452 fn drop(&mut self) {453 unsafe { fetchers_settings_free(self.0) };454 }455}456pub struct FlakeSettings(*mut flake_settings);457impl FlakeSettings {458 pub fn new() -> Result<Self> {459 with_default_context(|c, _| unsafe { flake_settings_new(c) }).map(Self)460 }461}462unsafe impl Send for FlakeSettings {}463unsafe impl Sync for FlakeSettings {}464impl Drop for FlakeSettings {465 fn drop(&mut self) {466 unsafe {467 flake_settings_free(self.0);468 }469 }470}471472pub struct FlakeReferenceParseFlags(*mut flake_reference_parse_flags);473impl FlakeReferenceParseFlags {474 pub fn new(settings: &FlakeSettings) -> Result<Self> {475 with_default_context(|c, _| unsafe { flake_reference_parse_flags_new(c, settings.0) })476 .map(Self)477 }478 pub fn set_base_dir(&mut self, dir: &str) -> Result<()> {479 with_default_context(|c, _| {480 unsafe {481 flake_reference_parse_flags_set_base_directory(482 c,483 self.0,484 dir.as_ptr().cast(),485 dir.len(),486 )487 };488 })489 }490}491impl Drop for FlakeReferenceParseFlags {492 fn drop(&mut self) {493 unsafe {494 flake_reference_parse_flags_free(self.0);495 }496 }497}498pub struct FlakeLockFlags(*mut flake_lock_flags);499impl FlakeLockFlags {500 pub fn new(settings: &FlakeSettings) -> Result<Self> {501 let o = with_default_context(|c, _| unsafe { flake_lock_flags_new(c, settings.0) })502 .map(Self)?;503 // with_default_context(|c, _| unsafe { flake_lock_flags_set_mode_virtual(c, o.0) })?;504505 Ok(o)506 }507}508impl Drop for FlakeLockFlags {509 fn drop(&mut self) {510 unsafe {511 flake_lock_flags_free(self.0);512 }513 }514}515516pub(crate) unsafe extern "C" fn copy_nix_str(517 start: *const c_char,518 n: c_uint,519 user_data: *mut c_void,520) {521 let s = unsafe { slice::from_raw_parts(start.cast::<u8>(), n as usize) };522 let s = std::str::from_utf8(s).expect("c string has invalid utf-8");523 unsafe { *user_data.cast::<String>() = s.to_owned() };524}525526pub struct Store(*mut c_store);527unsafe impl Send for Store {}528unsafe impl Sync for Store {}529530pub fn eval_store() -> Arc<Store> {531 GLOBAL_STATE.store.clone()532}533534impl Store {535 pub fn open(uri: &str) -> Result<Self> {536 let uri = CString::new(uri)?;537 let ptr = with_default_context(|c, _| unsafe { store_open(c, uri.as_ptr(), null_mut()) })?;538 if ptr.is_null() {539 bail!("failed to open store");540 }541 Ok(Store(ptr))542 }543544 pub fn parse_path(&self, path: &Utf8Path) -> Result<StorePath> {545 let path = CString::new(path.as_str()).expect("valid cstr");546 with_default_context(|c, _| {547 StorePath(unsafe { store_parse_path(c, self.0, path.as_ptr()) })548 })549 }550551 #[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();557558 if err.is_empty() {559 Ok(())560 } else {561 bail!("failed to sign {path}: {err}");562 }563 }564565 #[instrument(skip(self, dst))]566 pub fn copy_to(&self, dst: &Store, path: &Utf8Path) -> Result<()> {567 let sp = self568 .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 }578579 /// 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 }589590 #[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 }601602 #[instrument(skip(self, paths))]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 }611612 #[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 }616617 #[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 }631632 #[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 }642643 fn as_ptr(&self) -> *mut c_store {644 self.0645 }646}647impl Drop for Store {648 fn drop(&mut self) {649 unsafe { store_free(self.0) }650 }651}652653#[repr(transparent)]654pub struct EvalState(*mut c_eval_state);655unsafe impl Send for EvalState {}656unsafe impl Sync for EvalState {}657658impl Drop for EvalState {659 fn drop(&mut self) {660 unsafe {661 state_free(self.0);662 }663 }664}665666pub struct FlakeReference(*mut flake_reference);667impl FlakeReference {668 #[instrument(name = "new-flake-reference", skip(flake, parse, fetch))]669 pub fn new(670 s: &str,671 flake: &FlakeSettings,672 parse: &FlakeReferenceParseFlags,673 fetch: &FetchSettings,674 ) -> Result<(Self, String)> {675 let mut out = null_mut();676 let mut fragment = String::new();677 // let fetch_settings = fetcher_settings;678 with_default_context(|c, _| unsafe {679 flake_reference_and_fragment_from_string(680 c,681 fetch.0,682 flake.0,683 parse.0,684 s.as_ptr().cast(),685 s.len(),686 &mut out,687 Some(copy_nix_str),688 (&raw mut fragment).cast(),689 )690 })?;691 assert!(!out.is_null());692693 Ok((Self(out), fragment))694 }695 #[instrument(name = "lock-flake", skip(self, fetch, flake, lock))]696 pub fn lock(697 &mut self,698 fetch: &FetchSettings,699 flake: &FlakeSettings,700 lock: &FlakeLockFlags,701 ) -> Result<LockedFlake> {702 with_default_context(|c, es| unsafe { flake_lock(c, fetch.0, flake.0, es, lock.0, self.0) })703 .map(LockedFlake)704 }705}706unsafe impl Send for FlakeReference {}707unsafe impl Sync for FlakeReference {}708709pub struct LockedFlake(*mut locked_flake);710impl LockedFlake {711 pub fn get_attrs(&self, settings: &mut FlakeSettings) -> Result<Value> {712 with_default_context(|c, es| unsafe {713 locked_flake_get_output_attrs(c, settings.0, es, self.0)714 })715 .map(Value)716 }717}718unsafe impl Send for LockedFlake {}719unsafe impl Sync for LockedFlake {}720impl Drop for LockedFlake {721 fn drop(&mut self) {722 unsafe {723 locked_flake_free(self.0);724 };725 }726}727728type FieldName = [u8; 64];729fn init_field_name(v: &str) -> FieldName {730 let mut f = [0; 64];731 assert!(v.len() < 64, "max field name is 63 chars");732 assert!(733 v.bytes().all(|v| v != 0),734 "nul bytes are unsupported in field name"735 );736 f[0..v.len()].copy_from_slice(v.as_bytes());737 f738}739740pub struct RealisedString(*mut realised_string);741impl fmt::Debug for RealisedString {742 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {743 self.as_str().fmt(f)744 }745}746747impl RealisedString {748 pub fn as_str(&self) -> &str {749 let len = unsafe { realised_string_get_buffer_size(self.0) };750 let data: *const u8 = unsafe { realised_string_get_buffer_start(self.0) }.cast();751 let data = unsafe { slice::from_raw_parts(data, len) };752 std::str::from_utf8(data).expect("non-utf8 strings not supported")753 }754 pub fn path_count(&self) -> usize {755 unsafe { realised_string_get_store_path_count(self.0) }756 }757 pub fn path(&self, i: usize) -> String {758 assert!(i < self.path_count());759 let path = unsafe { realised_string_get_store_path(self.0, i) };760 let mut err_out = String::new();761 unsafe { store_path_name(path, Some(copy_nix_str), (&raw mut err_out).cast()) };762 err_out763 }764}765766unsafe impl Send for RealisedString {}767impl Drop for RealisedString {768 fn drop(&mut self) {769 unsafe { realised_string_free(self.0) }770 }771}772773#[repr(transparent)]774pub struct Value(*mut value);775776unsafe impl Send for Value {}777unsafe impl Sync for Value {}778779pub trait AsFieldName {780 fn as_field_name<T>(&self, v: impl FnOnce(FieldName) -> Result<T>) -> Result<T>;781 fn to_field_name(&self) -> Result<String>;782}783impl AsFieldName for Value {784 fn as_field_name<T>(&self, v: impl FnOnce(FieldName) -> Result<T>) -> Result<T> {785 let f = self.to_string()?;786 v(init_field_name(&f))787 }788 fn to_field_name(&self) -> Result<String> {789 self.to_string()790 }791}792impl<E> AsFieldName for E793where794 E: AsRef<str>,795{796 fn as_field_name<T>(&self, v: impl FnOnce(FieldName) -> Result<T>) -> Result<T> {797 let f = self.as_ref();798 v(init_field_name(f))799 }800 fn to_field_name(&self) -> Result<String> {801 Ok(self.as_ref().to_owned())802 }803}804805struct AttrsBuilder(*mut c_bindings_builder);806impl AttrsBuilder {807 fn new(capacity: usize) -> Self {808 with_default_context(|c, es| unsafe { make_bindings_builder(c, es, capacity) })809 .map(Self)810 .expect("alloc should not fail")811 }812 fn insert(&mut self, k: &impl AsFieldName, v: Value) {813 k.as_field_name(|name| {814 with_default_context(|c, _| unsafe {815 bindings_builder_insert(c, self.0, name.as_ptr().cast(), v.0);816 // bindings_builder_insert doesn't do incref817 })818 })819 .expect("builder insert shouldn't fail");820 }821}822impl Drop for AttrsBuilder {823 fn drop(&mut self) {824 unsafe { bindings_builder_free(self.0) };825 }826}827828struct ListBuilder(*mut c_list_builder, c_uint);829impl ListBuilder {830 fn new(capacity: usize) -> Self {831 with_default_context(|c, es| unsafe { make_list_builder(c, es, capacity) })832 .map(|l| Self(l, 0))833 .expect("alloc should not fail")834 }835}836impl ListBuilder {837 fn push(&mut self, v: Value) {838 with_default_context(|c, _| unsafe {839 list_builder_insert(840 c,841 self.0,842 {843 let v = self.1;844 self.1 += 1;845 v846 },847 v.0,848 )849 })850 .expect("list insert shouldn't fail");851 }852}853impl Drop for ListBuilder {854 fn drop(&mut self) {855 unsafe { list_builder_free(self.0) };856 }857}858859impl Value {860 pub fn new_primop(v: NativeFn) -> Self {861 let out = Self::new_uninit();862 with_default_context(|c, _| unsafe { init_primop(c, out.0, v.0) })863 .expect("primop initialization should not fail");864 out865 }866 pub fn new_attrs(v: HashMap<&str, Value>) -> Self {867 let out = Self::new_uninit();868 let mut b = AttrsBuilder::new(v.len());869 for (k, v) in v {870 b.insert(&k, v);871 }872 with_default_context(|c, _| unsafe { make_attrs(c, out.0, b.0) })873 .expect("attrs initialization should not fail");874875 out876 }877 fn new_list<T: Into<Self>>(v: Vec<T>) -> Self {878 let out = Self::new_uninit();879 let mut b = ListBuilder::new(v.len());880 for v in v {881 b.push(v.into());882 }883 with_default_context(|c, _| unsafe { make_list(c, b.0, out.0) })884 .expect("list initialization should not fail");885886 out887 }888 fn new_uninit() -> Self {889 let out = with_default_context(|c, es| unsafe { alloc_value(c, es) })890 .expect("value allocation should not fail");891 Self(out)892 }893 pub fn new_str(v: &str) -> Self {894 let s = CString::new(v).expect("string should not contain NULs");895 let out = Self::new_uninit();896 // String is copied, `s` is free to be dropped897 with_default_context(|c, _| unsafe { init_string(c, out.0, s.as_ptr()) })898 .expect("string initialization should not fail");899 out900 }901 pub fn new_int(i: i64) -> Self {902 let out = Self::new_uninit();903 with_default_context(|c, _| unsafe { init_int(c, out.0, i) })904 .expect("int initialization should not fail");905 out906 }907 pub fn new_bool(v: bool) -> Self {908 let out = Self::new_uninit();909 with_default_context(|c, _| unsafe { init_bool(c, out.0, v) })910 .expect("bool initialization should not fail");911 out912 }913 // TODO: As far as I can see, there is no way to get Thunks from nix public C api, so this function is useless914 // fn force(&mut self, st: &mut EvalState) -> Result<()> {915 // with_default_context(|c, _| unsafe { value_force(c, st.0, self.0) })?;916 // Ok(())917 // }918 pub fn type_of(&self) -> NixType {919 let ty = with_default_context(|c, _| unsafe { get_type(c, self.0) })920 .expect("get_type should not fail");921 NixType::from_int(ty)922 }923 fn builtin_to_string(&self) -> Result<Self> {924 let builtin = Self::eval("builtins.toString")?;925 builtin.call(self.clone())926 }927 fn force(&mut self, s: *mut nix_raw::EvalState) -> Result<()> {928 with_default_context(|c, _| unsafe { value_force(c, s, self.0) })?;929 Ok(())930 }931 pub fn to_string(&self) -> Result<String> {932 let ty = self.type_of();933 if !matches!(ty, NixType::String) {934 bail!("unexpected type: {ty:?}, expected string");935 }936 let mut str_out = String::new();937 with_default_context(|c, _| unsafe {938 get_string(c, self.0, Some(copy_nix_str), (&raw mut str_out).cast())939 })?;940941 Ok(str_out)942 }943 pub fn to_realised_string(&self) -> Result<RealisedString> {944 with_default_context(|c, es| unsafe { string_realise(c, es, self.0, false) })945 .map(RealisedString)946947 // let store_paths = unsafe { nix_raw::realised_string_get_store_path_count(str) };948 // for i in 0..store_paths {949 // let store_path = unsafe { nix_raw::realised_string_get_store_path(str, i) };950 // nix_raw::store_path_name(store_path, callback, user_data);951 // }952 // dbg!(store_paths);953 // todo!();954 }955956 pub fn has_field(&self, field: &str) -> Result<bool> {957 if !matches!(self.type_of(), NixType::Attrs) {958 bail!("invalid type: expected attrs");959 }960961 let f = init_field_name(field);962 with_default_context(|c, es| unsafe { has_attr_byname(c, self.0, es, f.as_ptr().cast()) })963 }964 // pub fn derivation_path(&self) {965 // nix_raw::real966 // }967 pub fn list_fields(&self) -> Result<Vec<String>> {968 if !matches!(self.type_of(), NixType::Attrs) {969 bail!("invalid type: expected attrs");970 }971972 let len = with_default_context(|c, _| unsafe { get_attrs_size(c, self.0) })?;973 let mut out = Vec::with_capacity(len as usize);974975 for i in 0..len {976 let name =977 with_default_context(|c, es| unsafe { get_attr_name_byidx(c, self.0, es, i) })?;978 let c = unsafe { CStr::from_ptr(name) };979 out.push(c.to_str().expect("nix field names are utf-8").to_owned());980 }981 Ok(out)982 }983 pub fn get_elem(&self, v: usize) -> Result<Self> {984 if !matches!(self.type_of(), NixType::List) {985 bail!("invalid type: expected list");986 }987 let len = with_default_context(|c, _| unsafe { get_list_size(c, self.0) })? as usize;988 if v >= len {989 bail!("oob list get: {v} >= {len}");990 }991992 with_default_context(|c, es| unsafe { get_list_byidx(c, self.0, es, v as u32) }).map(Self)993 }994 pub fn attrs_update(self, other: Value /*, ignore_errors: bool*/) -> Result<Self> {995 let attrs_update_fn = Self::eval("a: b: a // b")?;996997 attrs_update_fn998 .call(self)?999 .call(other)1000 .context("attrs update")1001 }1002 pub fn get_field(&self, name: impl AsFieldName) -> Result<Self> {1003 if !matches!(self.type_of(), NixType::Attrs) {1004 bail!("invalid type: expected attrs");1005 }10061007 name.as_field_name(|name| {1008 with_default_context(|c, es| unsafe {1009 get_attr_byname(c, self.0, es, name.as_ptr().cast())1010 })1011 .map(Self)1012 })1013 .with_context(|| format!("getting field {:?}", name.to_field_name()))1014 }1015 pub fn call(&self, v: Value) -> Result<Self> {1016 let kind = self1017 .functor_kind()1018 .ok_or_else(|| anyhow!("can only call function or functor"))?;10191020 let function = match kind {1021 FunctorKind::Function => self.clone(),1022 FunctorKind::Functor => {1023 let f = self1024 .get_field("__functor")1025 .context("getting functor value")?;1026 assert_eq!(1027 f.type_of(),1028 NixType::Function,1029 "invalid functor encountered"1030 );1031 f1032 }1033 };10341035 let out = Value::new_uninit();1036 with_default_context(|c, es| unsafe { value_call(c, es, function.0, v.0, out.0) })?;10371038 Ok(out)1039 }1040 pub fn eval(v: &str) -> Result<Self> {1041 let s = CString::new(v).expect("expression shouldn't have internal NULs");1042 let out = Self::new_uninit();1043 with_default_context(|c, es| unsafe {1044 expr_eval_from_string(c, es, s.as_ptr(), c"/root".as_ptr(), out.0)1045 })?;1046 Ok(out)1047 }1048 #[instrument(name = "build", skip(self), fields(output))]1049 pub fn build(&self, output: &str) -> Result<Utf8PathBuf> {1050 if !self.is_derivation() {1051 bail!("expected derivation to build")1052 }1053 let output_name = self1054 .get_field("outputName")1055 .context("getting output name field")?1056 .to_string()?;1057 let v = if output_name != output {1058 let out = self.get_field(output).context("getting target output")?;1059 if !out.is_derivation() {1060 bail!("unknown output: {output}");1061 }1062 out1063 } else {1064 self.clone()1065 };10661067 let drv_path = Utf8PathBuf::from(1068 v.get_field("drvPath")1069 .context("getting drvPath")?1070 .to_string()?,1071 );1072 let graph = Arc::new(drv::DrvGraph::resolve(&eval_store(), &drv_path)?);1073 let _guard = logging::register_build_graph(&Span::current(), &graph);10741075 scheduler::build_graph_sync(graph.clone(), vec![output.to_owned()])?;10761077 let s = v.builtin_to_string()?;1078 let rs = s.to_realised_string()?;1079 let out_path = rs.as_str().to_owned();1080 Ok(Utf8PathBuf::from(out_path))1081 }1082 pub fn as_json<T: DeserializeOwned>(&self) -> Result<T> {1083 let to_json = Self::eval("builtins.toJSON")?;1084 let s = to_json.call(self.clone())?.to_string()?;1085 Ok(serde_json::from_str(&s)?)1086 }1087 pub fn serialized<T: Serialize>(v: &T) -> Result<Self> {1088 Self::eval(&nixlike::serialize(v)?)1089 }10901091 // Convert to string/evaluate derivations/etc1092 // fn to_string_weak(&self) -> Result<String> {1093 // // TODO: For now, it works exactly like to_string, see the comment for fn force()1094 // self.to_string()1095 // }10961097 fn is_derivation(&self) -> bool {1098 if !matches!(self.type_of(), NixType::Attrs) {1099 return false;1100 }1101 let Some(ty) = self.get_field("type").ok() else {1102 return false;1103 };1104 matches!(ty.to_string().as_deref(), Ok("derivation"))1105 }1106 fn functor_kind(&self) -> Option<FunctorKind> {1107 match self.type_of() {1108 NixType::Attrs => self1109 .has_field("__functor")1110 .expect("has_field shouldn't fail for attrs")1111 .then_some(FunctorKind::Functor),1112 NixType::Function => Some(FunctorKind::Function),1113 _ => None,1114 }1115 }1116 pub fn is_function(&self) -> bool {1117 self.functor_kind().is_some()1118 }1119 pub fn is_null(&self) -> bool {1120 matches!(self.type_of(), NixType::Null)1121 }1122 pub fn is_string(&self) -> bool {1123 matches!(self.type_of(), NixType::String)1124 }1125 pub fn is_attrs(&self) -> bool {1126 matches!(self.type_of(), NixType::Attrs)1127 }1128}11291130impl From<String> for Value {1131 fn from(value: String) -> Self {1132 Value::new_str(&value)1133 }1134}1135impl From<bool> for Value {1136 fn from(value: bool) -> Self {1137 Value::new_bool(value)1138 }1139}1140impl From<&str> for Value {1141 fn from(value: &str) -> Self {1142 Value::new_str(value)1143 }1144}1145impl<T> From<Vec<T>> for Value1146where1147 T: Into<Value>,1148{1149 fn from(value: Vec<T>) -> Self {1150 Value::new_list(value)1151 }1152}11531154impl Clone for Value {1155 fn clone(&self) -> Self {1156 with_default_context(|c, _| unsafe { value_incref(c, self.0) })1157 .expect("value incref should not fail");1158 Self(self.0)1159 }1160}1161impl Drop for Value {1162 fn drop(&mut self) {1163 with_default_context(|c, _| unsafe { value_decref(c, self.0) })1164 .expect("value drop should not fail");1165 }1166}11671168static TOKIO_FOR_NIX: OnceLock<Arc<tokio::runtime::Runtime>> = OnceLock::new();11691170pub fn init_libraries() {1171 unsafe { GC_allow_register_threads() };11721173 let mut ctx = NixContext::new();1174 ctx.run_in_context(|c| unsafe { libutil_init(c) })1175 .expect("util init should not fail");1176 ctx.run_in_context(|c| unsafe { libstore_init(c) })1177 .expect("store init should not fail");1178 ctx.run_in_context(|c| unsafe { libexpr_init(c) })1179 .expect("expr init should not fail");11801181 nix_logging_cxx::apply_tracing_logger();1182}11831184pub fn init_tokio_for_nix(tokio: Arc<tokio::runtime::Runtime>) {1185 TOKIO_FOR_NIX1186 .set(tokio)1187 .expect("tokio for nix should only be initialized once");1188}11891190pub fn await_in_nix<F: Send + 'static>(f: impl Future<Output = F> + Send + 'static) -> F {1191 // It should be possible to do Handle::current(), but some of the planned features don't work well with that1192 let runtime = TOKIO_FOR_NIX1193 .get()1194 .expect("init_tokio_for_nix was not called");1195 std::thread::spawn(move || runtime.block_on(f))1196 .join()1197 .expect("await_in_nix inner thread panicked")1198}11991200unsafe extern "C" fn nix_primop_closure_adapter<const N: usize>(1201 user_data: *mut c_void,1202 mut context: *mut c_context,1203 state: *mut nix_raw::EvalState,1204 args: *mut *mut value,1205 ret: *mut value,1206) {1207 let user_closure: &UserClosure<N> = unsafe { &*user_data.cast_const().cast() };1208 let args: [&Value; N] = array::from_fn(|i| {1209 let v: &mut Value = unsafe { &mut *args.add(i).cast() };1210 v as &Value1211 });1212 let ctx: &mut NixContext = unsafe { transmute(&mut context) };12131214 let state: &EvalState = unsafe { std::mem::transmute(&state) };12151216 match user_closure(state, args) {1217 Ok(v) => {1218 unsafe { copy_value(context, ret, v.0) };1219 }1220 Err(e) => {1221 ctx.set_err(e);1222 }1223 }1224}12251226type UserClosure<const N: usize> = Box<dyn Fn(&EvalState, [&Value; N]) -> Result<Value>>;12271228pub struct NativeFn(*mut PrimOp);1229impl NativeFn {1230 pub fn new<const N: usize>(1231 name: &'static CStr,1232 doc: &'static CStr,1233 args: [&'static CStr; N],1234 f: impl Fn(&EvalState, [&Value; N]) -> Result<Value> + 'static,1235 ) -> Self {1236 // Double-boxing to make it thin pointer, as vtable gets outside of first Box1237 let closure: Box<UserClosure<N>> = Box::new(Box::new(f));1238 let f: PrimOpFun = Some(nix_primop_closure_adapter::<N>);1239 let mut args = args.into_iter().map(|v| v.as_ptr()).collect_vec();1240 args.push(null());1241 let args = args.as_mut_ptr();1242 let primop = unsafe {1243 alloc_primop(1244 null_mut(),1245 f,1246 N as i32,1247 name.as_ptr(),1248 args,1249 doc.as_ptr(),1250 Box::into_raw(closure).cast(),1251 )1252 };12531254 assert!(!primop.is_null(), "primop allocation should not fail");12551256 Self(primop)1257 }1258 pub fn register(self) {1259 unsafe { register_primop(null_mut(), self.0) };1260 }1261}12621263pub struct StorePath(*mut c_store_path);1264impl StorePath {1265 fn as_ptr(&self) -> *mut c_store_path {1266 self.01267 }1268}12691270impl Drop for StorePath {1271 fn drop(&mut self) {1272 unsafe { store_path_free(self.0) }1273 }1274}12751276#[test_log::test]1277fn test_native() -> Result<()> {1278 init_libraries();1279 NativeFn::new(1280 c"__uppercaseSuffix2",1281 c"make string uppercase and add suffix",1282 [c"str", c"suffix"],1283 |_, [str, suffix]: [&Value; 2]| {1284 let str = str.to_string()?;1285 let suffix = suffix.to_string()?;1286 Ok(Value::new_str(&format!("{}{suffix}", str.to_uppercase())))1287 },1288 )1289 .register();12901291 let mut fetch_settings = FetchSettings::new();1292 fetch_settings.set(c"warn-dirty", c"false");12931294 let manifest = format!("git+file://{}/../../", env!("CARGO_MANIFEST_DIR"));1295 let flake = FlakeSettings::new()?;1296 let parse = FlakeReferenceParseFlags::new(&flake)?;1297 let (mut r, _) = FlakeReference::new(&manifest, &flake, &parse, &fetch_settings)?;1298 let lock = FlakeLockFlags::new(&flake)?;1299 let locked = r.lock(&fetch_settings, &flake, &lock)?;1300 let attrs = locked.get_attrs(&mut FlakeSettings::new()?)?;13011302 let builtins = Value::eval("builtins")?;1303 assert_eq!(builtins.type_of(), NixType::Attrs);13041305 assert_eq!(attrs.type_of(), NixType::Attrs);1306 let test_data = nix_go!(attrs.testData);13071308 let test_string: String = nix_go_json!(test_data.testString);1309 assert_eq!(test_string, "hello");13101311 let s = nix_go!(attrs.packages["x86_64-linux"].fleet.drvPath);1312 let s = CString::new(s.to_string()?).expect("path str is cstring");13131314 let uppercase_suffix = Value::new_primop(NativeFn::new(1315 c"uppercase_suffix",1316 c"make string uppercase and add suffix",1317 [c"str", c"suffix"],1318 |es, [str, suffix]: [&Value; 2]| {1319 let str = str.to_string()?;1320 let suffix = suffix.to_string()?;1321 Ok(Value::new_str(&format!("{}{suffix}", str.to_uppercase())))1322 },1323 ));13241325 let test_result: String = nix_go_json!(test_data.testPrimop(uppercase_suffix));1326 assert_eq!(test_result, "PREFIX_BODY_SUFFIX");1327 let test_result: String = nix_go_json!(builtins.uppercaseSuffix2("test")("suffix"));1328 assert_eq!(test_result, "TESTsuffix");13291330 let drv_path =1331 nix_go!(attrs.packages["x86_64-linux"]["fleet-install-secrets"].drvPath).to_string()?;1332 let graph = drv::DrvGraph::resolve(&drv_path)?;1333 eprintln!(1334 "fleet-install-secrets dependency graph: {} nodes",1335 graph.nodes.len()1336 );1337 for (path, node) in &graph.nodes {1338 if !node.input_drvs.is_empty() {1339 eprintln!(" {} ({} deps)", node.name, node.input_drvs.len());1340 }1341 }13421343 Ok(())1344}13451346// pub struct GcAlloc;1347// unsafe impl GlobalAlloc for GcAlloc {1348// unsafe fn alloc(&self, l: Layout) -> *mut u8 {1349// let ptr = unsafe { GC_malloc(l.size()) };1350// ptr.cast()1351// }1352// unsafe fn dealloc(&self, ptr: *mut u8, _: Layout) {1353// // unsafe { GC_free(ptr.cast()) };1354// }1355//1356// unsafe fn realloc(&self, ptr: *mut u8, _: Layout, new_size: usize) -> *mut u8 {1357// let ptr = unsafe { GC_realloc(ptr.cast(), new_size) };1358// ptr.cast()1359// }1360// }1361//1362// #[global_allocator]1363// static GC: GcAlloc = GcAlloc;crates/nix-eval/src/logging.rsdiffbeforeafterboth--- a/crates/nix-eval/src/logging.rs
+++ b/crates/nix-eval/src/logging.rs
@@ -109,6 +109,12 @@
(ActivityType::CopyPaths, []) => {
debug_span!(target: "nix::copy-paths", "copying paths")
}
+ // Umbrella progress activity. Nix emits it at lvlError ("always show"),
+ // which would otherwise become a noisy ERROR `action` span — give it an
+ // explicit debug level like the other container activities.
+ (ActivityType::Builds, []) => {
+ debug_span!(target: "nix::builds", "building")
+ }
(ActivityType::Unknown, [])
if s.starts_with("copying \"") && s.ends_with("\" to the store") =>
{
@@ -365,6 +371,16 @@
BuildGraphGuard { paths }
}
+/// Drop the lazily-created `building` span for a derivation. Used when a build
+/// fails: its dependents will never build, so their speculatively-created spans
+/// (made when a dependency started building) would otherwise linger until the
+/// whole build graph is torn down.
+pub fn remove_drv_span(drv_path: &Utf8Path) {
+ if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(drv_path) {
+ entry.span = None;
+ }
+}
+
fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {
let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");
@@ -524,7 +540,7 @@
// ResultType::FileLinked => todo!(),
(ResultType::BuildLogLine, [Str(s)]) => {
let s = ansi_filter(s);
- info!("{s}");
+ info!(target: "nix::build", "{s}");
}
// ResultType::UntrustedPath => todo!(),
// ResultType::CorruptedPath => todo!(),
crates/nix-eval/src/scheduler.rsdiffbeforeafterboth--- a/crates/nix-eval/src/scheduler.rs
+++ b/crates/nix-eval/src/scheduler.rs
@@ -45,9 +45,36 @@
},
}
+// Derivations whose names contain any of these are "heavy": memory-hungry builds that already
+// saturate all cores internally (e.g. composable-kernels), so running anything alongside them
+// just thrashes RAM. A heavy build grabs every semaphore permit and runs exclusively.
+const DEFAULT_HEAVY_DRVS: &[&str] = &[
+ "composable-kernel",
+ "composable_kernel",
+ "ghc",
+ "gfortran",
+ "llvm",
+ "clang",
+];
+
+fn heavy_patterns_from_env() -> Vec<String> {
+ let mut pats: Vec<String> = DEFAULT_HEAVY_DRVS.iter().map(|s| (*s).to_owned()).collect();
+ if let Ok(extra) = std::env::var("FLEET_HEAVY_DRVS") {
+ pats.extend(
+ extra
+ .split(',')
+ .map(str::trim)
+ .filter(|s| !s.is_empty())
+ .map(str::to_owned),
+ );
+ }
+ pats
+}
+
pub struct Scheduler {
store: Arc<Store>,
parallelism: usize,
+ heavy_patterns: Vec<String>,
events: broadcast::Sender<BuildEvent>,
}
@@ -58,10 +85,20 @@
Self {
store: eval_store(),
parallelism,
+ heavy_patterns: heavy_patterns_from_env(),
events,
}
}
+ // Permits a drv must hold to run. Heavy drvs take all of them, pausing everything else.
+ fn permits_for(&self, name: &str) -> u32 {
+ if self.heavy_patterns.iter().any(|p| name.contains(p)) {
+ self.parallelism as u32
+ } else {
+ 1
+ }
+ }
+
pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {
self.events.subscribe()
}
@@ -139,6 +176,7 @@
.get(&path)
.map(|n| n.name.clone())
.unwrap_or_default();
+ crate::logging::remove_drv_span(&path);
let _ = self.events.send(BuildEvent::DrvCancelled {
drv_path: path.clone(),
name,
@@ -153,8 +191,21 @@
let graph = graph.clone();
let wanted_here = wanted.get(&path).cloned().unwrap_or_default();
let store = self.store.clone();
+ let permits = graph
+ .nodes
+ .get(&path)
+ .map(|n| self.permits_for(&n.name))
+ .unwrap_or(1);
in_flight.push(tokio::spawn(async move {
- let _permit = sem.acquire_owned().await.expect("semaphore not closed");
+ if permits > 1 {
+ debug!(permits, "heavy drv: waiting for exclusive build slot");
+ }
+ // Heavy drvs acquire every permit, so they only start once all in-flight
+ // builds drain and nothing new can slip in (the semaphore is FIFO-fair).
+ let _permit = sem
+ .acquire_many_owned(permits)
+ .await
+ .expect("semaphore not closed");
let node = graph
.nodes
.get(&path)
@@ -224,6 +275,7 @@
propagate_done(&dependents, &mut indeg, &mut ready, &finished);
}
Err(e) => {
+ crate::logging::remove_drv_span(&finished);
failed.insert(finished.clone(), format!("{e:#}"));
mark_tainted(&dependents, &finished, &mut tainted);
propagate_done(&dependents, &mut indeg, &mut ready, &finished);
@@ -368,11 +420,23 @@
// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing
// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one
// will work as a hard cap.
+/// Number of derivations to build concurrently, taken from Nix's `max-jobs`
+/// setting. `auto` resolves to the CPU count, matching Nix itself.
+fn max_jobs() -> usize {
+ let cores = || {
+ std::thread::available_parallelism()
+ .map(|p| p.get())
+ .unwrap_or(4)
+ };
+ match crate::get_setting(c"max-jobs") {
+ Ok(v) if v.trim() == "auto" => cores(),
+ Ok(v) => v.trim().parse().unwrap_or_else(|_| cores()),
+ Err(_) => cores(),
+ }
+}
+
pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {
- let parallelism = std::thread::available_parallelism()
- .map(|p| p.get())
- .unwrap_or(4);
- let scheduler = Scheduler::new(parallelism);
+ let scheduler = Scheduler::new(max_jobs());
crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })
.context("scheduler run")
}
crates/remowt-fleet/src/lib.rsdiffbeforeafterboth--- a/crates/remowt-fleet/src/lib.rs
+++ b/crates/remowt-fleet/src/lib.rs
@@ -28,6 +28,8 @@
Sign(String),
#[error("listing generations failed: {0}")]
ListGenerations(String),
+ #[error("substitution failed: {0}")]
+ Substitute(String),
}
#[endpoints(ns = 91)]
@@ -60,6 +62,20 @@
.map_err(|e| NixError::Sign(e.to_string()))
}
+ /// Download the given paths (and their closures) into the local store using
+ /// the configured substituters. Used to fetch closures that were uploaded to
+ /// a binary cache (e.g. attic) instead of copied over the nix daemon tunnel.
+ #[endpoints(id = 6)]
+ async fn substitute(&self, paths: Vec<Utf8PathBuf>) -> Result<Vec<Utf8PathBuf>, NixError> {
+ spawn_blocking(move || {
+ let store = eval_store();
+ store.substitute_paths(&paths)
+ })
+ .await
+ .expect("substitution panicked")
+ .map_err(|e| NixError::Substitute(e.to_string()))
+ }
+
#[endpoints(id = 5)]
async fn list_generations(
&self,
modules/cache.nixdiffbeforeafterboth--- /dev/null
+++ b/modules/cache.nix
@@ -0,0 +1,61 @@
+{
+ lib,
+ fleetLib,
+ config,
+ ...
+}:
+let
+ inherit (lib.options) mkOption mkEnableOption literalExpression;
+ inherit (lib.types)
+ enum
+ str
+ bool
+ submodule
+ ;
+ inherit (fleetLib.options) mkHostsOption;
+
+ _file = ./cache.nix;
+
+ cacheOptions = {
+ enable = mkEnableOption "fleet binary cache";
+ kind = mkOption {
+ description = ''
+ Kind of the binary cache to use.
+ '';
+ type = enum [ "attic" ];
+ default = "attic";
+ };
+ name = mkOption {
+ description = ''
+ Name of the binary cache.
+ '';
+ type = str;
+ };
+ pin = mkOption {
+ description = ''
+ Whether to pin pushed paths in the attic cache, preventing them from
+ being garbage collected.
+
+ Requires compatible fork of attic server+client: https://github.com/zhaofengli/attic/pull/227#issuecomment-4759154796
+ '';
+ type = bool;
+ default = false;
+ };
+ };
+in
+{
+ options = {
+ cache = cacheOptions;
+ hosts = mkHostsOption {
+ inherit _file;
+ options.cache = mkOption {
+ description = ''
+ Binary cache configuration for this host.
+ '';
+ type = submodule { options = cacheOptions; };
+ default = config.cache;
+ defaultText = literalExpression "fleetConfiguration.cache";
+ };
+ };
+ };
+}
modules/module-list.nixdiffbeforeafterboth--- a/modules/module-list.nix
+++ b/modules/module-list.nix
@@ -1,5 +1,6 @@
[
./assertions.nix
+ ./cache.nix
./fleetLib.nix
./hosts.nix
./nixos.nix
pkgs/default.nixdiffbeforeafterboth--- a/pkgs/default.nix
+++ b/pkgs/default.nix
@@ -12,8 +12,6 @@
fleet-generator-helper = callPackage ./fleet-generator-helper.nix { inherit craneLib; };
inherit remowt-agents-bundle;
- remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix {
- inherit craneLib inputs remowt-agents-bundle;
- };
+ remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix { inherit craneLib inputs; };
remowt-ssh = callPackage ./remowt-ssh.nix { inherit craneLib remowt-agents-bundle; };
}
remowt/cmds/remowt-agent/src/main.rsdiffbeforeafterboth--- a/remowt/cmds/remowt-agent/src/main.rs
+++ b/remowt/cmds/remowt-agent/src/main.rs
@@ -273,10 +273,11 @@
let helper = SocketHelper {
fallback: SuidHelper,
};
- register_auth_agent(&system_conn, Agent::new(helper, RofiPrompter)).await?;
+ let rofi = RofiPrompter::default();
+ register_auth_agent(&system_conn, Agent::new(helper, rofi.clone())).await?;
let mut rpc = Rpc::<BifConfig>::new(Address::User);
- serve_prompts(&mut rpc, RofiPrompter);
+ serve_prompts(&mut rpc, rofi);
gateway::serve(rpc.clone(), &gateway::local_socket()?).await?;
remowt/cmds/remowt-ssh/Cargo.tomldiffbeforeafterboth--- a/remowt/cmds/remowt-ssh/Cargo.toml
+++ b/remowt/cmds/remowt-ssh/Cargo.toml
@@ -8,6 +8,7 @@
[dependencies]
clap = { workspace = true, features = ["derive"] }
tracing-subscriber.workspace = true
+tracing-journald.workspace = true
remowt-link-shared.workspace = true
remowt-client.workspace = true
tokio = { workspace = true, features = [
remowt/cmds/remowt-ssh/src/main.rsdiffbeforeafterboth--- a/remowt/cmds/remowt-ssh/src/main.rs
+++ b/remowt/cmds/remowt-ssh/src/main.rs
@@ -20,6 +20,8 @@
use tokio::io::{AsyncRead, ReadBuf};
use tokio::signal::unix::{signal, SignalKind};
use tracing::debug;
+use tracing_subscriber::prelude::*;
+use tracing_subscriber::EnvFilter;
#[derive(Parser)]
enum Opts {
@@ -45,10 +47,16 @@
#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> {
- tracing_subscriber::fmt()
- .with_writer(std::io::stderr)
- .without_time()
- .init();
+ // Log to the journal, not stderr: this is an interactive client that puts
+ // the terminal in raw mode, so anything on stderr corrupts the session
+ // output. If the journal socket is unavailable, drop logs rather than fall
+ // back to stderr.
+ if let Ok(journald) = tracing_journald::layer() {
+ tracing_subscriber::registry()
+ .with(EnvFilter::from_default_env())
+ .with(journald.with_syslog_identifier("remowt-ssh".to_owned()))
+ .init();
+ }
let opts = Opts::parse();
let bundle = AgentBundle::from_dir(agents_dir()?)?;
remowt/crates/remowt-ui-prompt/Cargo.tomldiffbeforeafterboth--- a/remowt/crates/remowt-ui-prompt/Cargo.toml
+++ b/remowt/crates/remowt-ui-prompt/Cargo.toml
@@ -12,5 +12,5 @@
remowt-link-shared.workspace = true
serde.workspace = true
thiserror.workspace = true
-tokio = { workspace = true, features = ["io-util", "macros", "process", "rt"] }
+tokio = { workspace = true, features = ["io-util", "macros", "process", "rt", "sync"] }
tracing.workspace = true
remowt/crates/remowt-ui-prompt/src/auto.rsdiffbeforeafterboth--- a/remowt/crates/remowt-ui-prompt/src/auto.rs
+++ b/remowt/crates/remowt-ui-prompt/src/auto.rs
@@ -24,7 +24,7 @@
};
Self {
remote,
- fallback: RofiPrompter,
+ fallback: RofiPrompter::default(),
}
}
remowt/crates/remowt-ui-prompt/src/rofi.rsdiffbeforeafterboth--- a/remowt/crates/remowt-ui-prompt/src/rofi.rs
+++ b/remowt/crates/remowt-ui-prompt/src/rofi.rs
@@ -1,13 +1,18 @@
use std::process::Stdio;
+use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
+use tokio::sync::Mutex;
use tracing::trace;
use crate::{Error, Prompter, Result, Source};
-#[derive(Clone)]
-pub struct RofiPrompter;
+#[derive(Clone, Default)]
+pub struct RofiPrompter {
+ // Rofi can't run concurrently; serialize invocations.
+ lock: Arc<Mutex<()>>,
+}
fn fixup_prompt(prompt: &str) -> &str {
// Rofi always appends such suffix
@@ -27,6 +32,7 @@
source: &[Source],
) -> Result<u32> {
trace!("rofi radio");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mesg = if source.is_empty() {
description.to_owned()
@@ -113,6 +119,7 @@
source: &[Source],
) -> Result<String> {
trace!("rofi text");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mesg = if source.is_empty() {
description.to_owned()
@@ -162,6 +169,7 @@
async fn display_text(&self, error: bool, description: &str, source: &[Source]) -> Result<()> {
trace!("rofi display");
+ let _guard = self.lock.lock().await;
let mut cmd = rofi_command();
let mut mesg = if source.is_empty() {
description.to_owned()
@@ -206,7 +214,7 @@
#[ignore = "interactive"]
async fn test() {
let prompter = PrependSourcePrompter {
- prompter: RofiPrompter,
+ prompter: RofiPrompter::default(),
description: "test".to_owned(),
source: vec![Source(Cow::Borrowed("ssh"))],
};