git.delta.rocks / fleet / refs/commits / 78196e185cf1

difftreelog

feat attic

ukmvyzywYaroslav Bolyukin2026-06-19parent: #2438e2b.patch.diff

22 files changed

modifiedCargo.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"
modifiedCargo.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
modifiedcmds/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),
addedcmds/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());
+	}
+}
modifiedcmds/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<()> {
modifiedcrates/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);
modifiedcrates/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,
modifiedcrates/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);
modifiedcrates/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")?;
 
modifiedcrates/nix-eval/src/lib.rsdiffbeforeafterboth
before · crates/nix-eval/src/lib.rs
1use 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;
modifiedcrates/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!(),
modifiedcrates/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")
 }
modifiedcrates/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,
addedmodules/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";
+      };
+    };
+  };
+}
modifiedmodules/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
modifiedpkgs/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; };
 }
modifiedremowt/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?;
 
modifiedremowt/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 = [
modifiedremowt/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()?)?;
modifiedremowt/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
modifiedremowt/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(),
 		}
 	}
 
modifiedremowt/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"))],
 		};