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

5084 "remowt-ui-prompt",5084 "remowt-ui-prompt",
5085 "tokio",5085 "tokio",
5086 "tracing",5086 "tracing",
5087 "tracing-journald",
5087 "tracing-subscriber",5088 "tracing-subscriber",
5088]5089]
50895090
6575 "tracing-subscriber",6576 "tracing-subscriber",
6576]6577]
6578
6579[[package]]
6580name = "tracing-journald"
6581version = "0.3.2"
6582source = "registry+https://github.com/rust-lang/crates.io-index"
6583checksum = "2d3a81ed245bfb62592b1e2bc153e77656d94ee6a0497683a65a12ccaf2438d0"
6584dependencies = [
6585 "libc",
6586 "tracing-core",
6587 "tracing-subscriber",
6588]
65776589
6578[[package]]6590[[package]]
6579name = "tracing-log"6591name = "tracing-log"

75tokio = { version = "1.45.1", features = ["fs", "macros", "rt", "rt-multi-thread", "sync", "time"] }75tokio = { version = "1.45.1", features = ["fs", "macros", "rt", "rt-multi-thread", "sync", "time"] }
76tracing = "0.1"76tracing = "0.1"
77tracing-indicatif = "0.3.13"77tracing-indicatif = "0.3.13"
78tracing-journald = "0.3.2"
78tracing-opentelemetry = "0.33.0"79tracing-opentelemetry = "0.33.0"
79uuid = { version = "1", features = ["v4"] }80uuid = { version = "1", features = ["v4"] }
80# For fixed coloring needs to be updated to https://github.com/tokio-rs/tracing/pull/348481# For fixed coloring needs to be updated to https://github.com/tokio-rs/tracing/pull/3484

14use tokio::task::spawn_blocking;14use tokio::task::spawn_blocking;
15use tracing::{Instrument, error, field, info, info_span, warn};15use tracing::{Instrument, error, field, info, info_span, warn};
16
17const TARGET: &str = "build-systems";
1618
17#[derive(Parser)]19#[derive(Parser)]
18pub struct Deploy {20pub struct Deploy {
32}34}
3335
34async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<Utf8PathBuf> {36async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<Utf8PathBuf> {
35	info!("building");37	info!(target: TARGET, "building");
36	let host = config.host(&hostname)?;38	let host = config.host(&hostname)?;
37	// let action = Action::from(self.subcommand.clone());39	// let action = Action::from(self.subcommand.clone());
38	let nixos = host.nixos_config()?;40	let nixos = host.nixos_config()?;
4345
44	// We already have system profiles for backups.46	// We already have system profiles for backups.
45	if !host.local {47	if !host.local {
46		info!("adding gc root");48		info!(target: TARGET, "adding gc root");
47		let local = config.local_host();49		let local = config.local_host();
48		let plugin_id = local.ensure_nix_plugin().await?;50		let plugin_id = local.ensure_nix_plugin().await?;
49		let nix = local51		let nix = local
70		let build_attr = self.build_attr.clone();72		let build_attr = self.build_attr.clone();
71		for host in hosts {73		for host in hosts {
72			let config = config.clone();74			let config = config.clone();
73			let span = info_span!("build", host = field::display(&host.name));75			let span = info_span!(target: TARGET, "build", host = field::display(&host.name));
74			let hostname = host.name;76			let hostname = host.name;
75			let build_attr = build_attr.clone();77			let build_attr = build_attr.clone();
76			tasks.push(78			tasks.push(
77				(async move {79				(async move {
78					let built = match build_task(config, hostname.clone(), &build_attr).await {80					let built = match build_task(config, hostname.clone(), &build_attr).await {
79						Ok(path) => path,81						Ok(path) => path,
80						Err(e) => {82						Err(e) => {
81							error!("failed to deploy host: {:?}", e);83							error!(target: TARGET, "failed to deploy host: {:?}", e);
82							return;84							return;
83						}85						}
84					};86					};
85					// TODO: Handle error87					// TODO: Handle error
86					let mut out = current_dir().expect("cwd exists");88					let mut out = current_dir().expect("cwd exists");
87					out.push(format!("built-{hostname}"));89					out.push(format!("built-{hostname}"));
8890
89					info!("linking iso image to {:?}", out);91					info!(target: TARGET, "linking iso image to {:?}", out);
90					if let Err(e) = symlink(built, out) {92					if let Err(e) = symlink(built, out) {
91						error!("failed to symlink: {e}")93						error!(target: TARGET, "failed to symlink: {e}")
92					}94					}
93				})95				})
94				.instrument(span),96				.instrument(span),
105		let tasks = FuturesUnordered::new();107		let tasks = FuturesUnordered::new();
106		for host in hosts.into_iter() {108		for host in hosts.into_iter() {
107			let config = config.clone();109			let config = config.clone();
108			let span = info_span!("deploy", host = field::display(&host.name));110			let span = info_span!(target: TARGET, "deploy", host = field::display(&host.name));
109			let hostname = host.name.clone();111			let hostname = host.name.clone();
110			let opts = opts.clone();112			let opts = opts.clone();
111			if let Some(deploy_kind) = opts.action_attr::<DeployKind>(&host, "deploy_kind")? {113			if let Some(deploy_kind) = opts.action_attr::<DeployKind>(&host, "deploy_kind")? {
125					{127					{
126						Ok(path) => path,128						Ok(path) => path,
127						Err(e) => {129						Err(e) => {
128							error!("failed to build host system closure: {:?}", e);130							error!(target: TARGET, "failed to build host system closure: {:?}", e);
129							return;131							return;
130						}132						}
131					};133					};
132134
133					let deploy_kind = match host.deploy_kind().await {135					let deploy_kind = match host.deploy_kind().await {
134						Ok(v) => v,136						Ok(v) => v,
135						Err(e) => {137						Err(e) => {
136							error!("failed to query target deploy kind: {e}");138							error!(target: TARGET, "failed to query target deploy kind: {e}");
137							return;139							return;
138						}140						}
139					};141					};
140142
141					// TODO: Make disable_rollback a host attribute instead143					// TODO: Make disable_rollback a host attribute instead
142					let mut disable_rollback = self.disable_rollback;144					let mut disable_rollback = self.disable_rollback;
143					if !disable_rollback && deploy_kind != DeployKind::Fleet {145					if !disable_rollback && deploy_kind != DeployKind::Fleet {
144						warn!("disabling rollback, as not supported by non-fleet deployment kinds");146						warn!(target: TARGET, "disabling rollback, as not supported by non-fleet deployment kinds");
145						disable_rollback = true;147						disable_rollback = true;
146					}148					}
147149
148					let remote_path =150					let remote_path =
149						match upload_task(&host, GenerationStorage::Deployer, built).await {151						match upload_task(&host, GenerationStorage::Deployer, built).await {
150							Ok(v) => v,152							Ok(v) => v,
151							Err(e) => {153							Err(e) => {
152								error!("upload failed: {e}");154								error!(target: TARGET, "upload failed: {e}");
153								return;155								return;
154							}156							}
155						};157						};
169					)171					)
170					.await172					.await
171					{173					{
172						error!("activation failed: {e}");174						error!(target: TARGET, "activation failed: {e}");
173					}175					}
174				})176				})
175				.instrument(span),177				.instrument(span),

1#![recursion_limit = "512"]1#![recursion_limit = "512"]
22
3pub(crate) mod cmds;3pub(crate) mod cmds;
4mod log_tree;
45
5use std::{process::ExitCode, sync::Arc};6use std::{process::ExitCode, sync::Arc};
67
21use human_repr::HumanCount;22use human_repr::HumanCount;
22#[cfg(feature = "indicatif")]23#[cfg(feature = "indicatif")]
23use indicatif::{ProgressState, ProgressStyle};24use indicatif::{ProgressState, ProgressStyle};
25use log_tree::TreeLayer;
24use nix_eval::{26use nix_eval::{
25	eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,27	eval_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
26};28};
128fn setup_logging(opts: &RootOpts) -> Result<()> {130fn setup_logging(opts: &RootOpts) -> Result<()> {
129	#[cfg(feature = "indicatif")]131	#[cfg(feature = "indicatif")]
130	let indicatif_layer = {132	let indicatif_layer = {
133		use std::fmt;
131		use std::time::Duration;134		use std::time::Duration;
132135
133		IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))136		IndicatifLayer::new().with_max_progress_bars(10, Some(ProgressStyle::default_spinner()))
137			.with_span_child_prefix_indent("  ")
138			.with_span_child_prefix_symbol("")
134			.with_progress_style(139			.with_progress_style(
135			ProgressStyle::with_template(140			ProgressStyle::with_template(
136				"{color_start}{span_child_prefix} {span_name}{{{span_fields}}}{color_end} {wide_msg} {color_start}{download_progress} {elapsed}{color_end}",141				"{span_child_prefix:.magenta}{chevron:.magenta}{span_name:<18.blue.bold} {wide_msg} {span_fields:.dim} {color_start}{download_progress} {elapsed}{color_end}",
137			)142			)
143				.unwrap()
138				.unwrap()144				.with_key("chevron", |_: &ProgressState, writer: &mut dyn fmt::Write| {
145					let _ = write!(writer, "> ");
146				})
139				.with_key("download_progress", |state: &ProgressState, writer: &mut dyn std::fmt::Write| {147				.with_key("download_progress", |state: &ProgressState, writer: &mut dyn fmt::Write| {
140					let Some(len) = state.len() else {148					let Some(len) = state.len() else {
141						return;149						return;
142					};150					};
149				})157				})
150				.with_key(158				.with_key(
151					"color_start",159					"color_start",
152					|state: &ProgressState, writer: &mut dyn std::fmt::Write| {160					|state: &ProgressState, writer: &mut dyn fmt::Write| {
153						let elapsed = state.elapsed();161						let elapsed = state.elapsed();
154162
155						if elapsed > Duration::from_secs(60) {163						if elapsed > Duration::from_secs(60) {
163				)171				)
164				.with_key(172				.with_key(
165					"color_end",173					"color_end",
166					|state: &ProgressState, writer: &mut dyn std::fmt::Write| {174					|state: &ProgressState, writer: &mut dyn fmt::Write| {
167						if state.elapsed() > Duration::from_secs(30) {175						if state.elapsed() > Duration::from_secs(30) {
168							let _ = write!(writer, "\x1b[0m");176							let _ = write!(writer, "\x1b[0m");
169						}177						}
174182
175	let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));183	let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
176184
177	let reg = tracing_subscriber::registry().with({185	let tree = {
178		let sub = tracing_subscriber::fmt::layer().without_time();186		#[cfg(feature = "indicatif")]
187		let writer = indicatif_layer.get_stderr_writer();
179		#[cfg(feature = "indicatif")]188		#[cfg(not(feature = "indicatif"))]
180		let sub = sub.with_writer(indicatif_layer.get_stderr_writer());189		let writer = || std::io::stderr();
181		sub.with_filter(filter) // .without,190		TreeLayer::new(writer)
182	});191	};
192
193	let reg = tracing_subscriber::registry().with(tree.with_filter(filter));
183194
184	#[cfg(feature = "indicatif")]195	#[cfg(feature = "indicatif")]
185	let reg = reg.with(indicatif_layer);196	let reg = reg.with(indicatif_layer);
200		let logger = OpenTelemetryTracingBridge::new(&log_provider);211		let logger = OpenTelemetryTracingBridge::new(&log_provider);
201		let tracer = span_provider.tracer("fleet");212		let tracer = span_provider.tracer("fleet");
202213
203		let reg = reg214		reg.with(tracing_opentelemetry::layer().with_tracer(tracer))
204			.with(tracing_opentelemetry::layer().with_tracer(tracer))
205			.with(logger);215			.with(logger)
206
207		reg.init();216			.init();
208	} else {217	} else {
209		reg.init();218		reg.init();
210	};219	}
211220
212	Ok(())221	Ok(())
213}222}
252		.await261		.await
253		.expect("primary task panicked")262		.expect("primary task panicked")
254	})263	})
255	// async_main(opts)
256}264}
257265
258async fn main_real(opts: RootOpts) -> Result<()> {266async fn main_real(opts: RootOpts) -> Result<()> {

290	if !host.local {290	if !host.local {
291		info!("uploading system closure");291		info!("uploading system closure");
292		let mut tries = 0;292		let mut tries = 0;
293		// TODO: Use gc prefix?..
293		loop {294		loop {
295			let name = format!("{}-system", host.gc_root_prefix());
294			match host.remote_derivation(&generation).await {296			match host.remote_derivation(&generation, Some(&name)).await {
295				Ok(remote) => {297				Ok(remote) => {
296					assert!(remote == generation, "CA derivations aren't implemented");298					assert!(remote == generation, "CA derivations aren't implemented");
297					return Ok(remote);299					return Ok(remote);

13use camino::{Utf8Path, Utf8PathBuf};13use camino::{Utf8Path, Utf8PathBuf};
14use chrono::{DateTime, Utc};14use chrono::{DateTime, Utc};
15use fleet_shared::SecretData;15use fleet_shared::SecretData;
16use nix_eval::{Store, Value, drv::DrvGraph, eval_store, nix_go, nix_go_json, util::assert_warn};16use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};
17use remowt_client::{AgentBundle, Remowt};17use remowt_client::{AgentBundle, Remowt};
18use remowt_endpoints::fs::FsClient;18use remowt_endpoints::fs::FsClient;
19use remowt_fleet::NixClient;19use remowt_fleet::NixClient;
20use remowt_link_shared::BifConfig;20use remowt_link_shared::{Address, BifConfig};
21use remowt_ui_prompt::auto::AutoPrompter;21use remowt_ui_prompt::auto::AutoPrompter;
22use remowt_ui_prompt::bifrost::PromptEndpoints;22use remowt_ui_prompt::bifrost::PromptEndpoints;
23use remowt_ui_prompt::{PrependSourcePrompter, Source};23use remowt_ui_prompt::{PrependSourcePrompter, Source};
24use tempfile::NamedTempFile;24use tempfile::NamedTempFile;
25use tokio::{sync::OnceCell, task::spawn_blocking};25use tokio::{process::Command, sync::OnceCell, task::spawn_blocking};
26use tracing::{info, instrument, warn};26use tracing::{info, instrument, warn};
2727
28use crate::fleetdata::{28use crate::fleetdata::{
104	session_destination: OnceLock<String>,104	session_destination: OnceLock<String>,
105	legacy_ssh_store: OnceLock<bool>,105	legacy_ssh_store: OnceLock<bool>,
106106
107	/// fleetConfiguration.hosts.host, None for local (as local is only used for local tool overrides)
107	pub host_config: Option<Value>,108	pub host_config: Option<Value>,
108	pub nixos_config: OnceLock<Value>,109	pub nixos_config: OnceLock<Value>,
109	pub nixos_unchecked_config: OnceLock<Value>,110	pub nixos_unchecked_config: OnceLock<Value>,
110	pub pkgs_override: Option<Value>,111	pub pkgs_override: Option<Value>,
111112
113	binary_cache: OnceCell<Option<BinaryCache>>,
114
112	// TODO: Move command helpers away with connectivity refactor115	// TODO: Move command helpers away with connectivity refactor
113	pub local: bool,116	pub local: bool,
114	pub remowt: OnceCell<Remowt>,117	pub remowt: OnceCell<Remowt>,
131	AgentBundle::from_dir(agents_dir()?)134	AgentBundle::from_dir(agents_dir()?)
132}135}
133136
134enum RemoteDerivationMode {
135	/// Closure is (pre)signed, remote host is expected to trust our signatures.137#[derive(Clone)]
136	CopySigned,
137	/// Closure is not signed, using escalated nix daemon to bypass signature verification.138pub enum BinaryCache {
138	CopyPrivileged,139	Attic { name: String, pin: bool },
139	/// Closure is uploaded to attic, remote host is expected to trust its signature.
140	Attic { name: String },
141}140}
141impl BinaryCache {
142	async fn upload(&self, local: &Remowt, path: &Utf8Path, pin_name: Option<&str>) -> Result<()> {
143		match self {
144			BinaryCache::Attic { name, pin } => {
145				let mut cmd = local.cmd("attic");
146				cmd.arg("push").arg(name).arg(path.as_str());
147				cmd.run()
148					.await
149					.context("failed to push path to attic (is attic-cli on PATH?)")?;
142150
151				if *pin && let Some(pin_name) = pin_name {
152					let mut cmd = local.cmd("attic");
153					cmd.arg("pin")
154						.arg("create")
155						.arg(name)
156						.arg(pin_name)
157						.arg(path.as_str());
158					cmd.run()
159						.await
160						.context("failed to pin path in attic (is your attic-cli compatible?)")?;
161				}
162				Ok(())
163			}
164		}
165	}
166}
167
143impl ConfigHost {168impl ConfigHost {
144	pub fn set_session_destination(&self, dest: String) {169	pub fn set_session_destination(&self, dest: String) {
145		self.session_destination170		self.session_destination
156			.set(legacy)181			.set(legacy)
157			.expect("legacy ssh store is already set")182			.expect("legacy ssh store is already set")
158	}183	}
184	pub fn binary_cache(&self) -> Result<Option<BinaryCache>> {
185		if let Some(v) = self.binary_cache.get() {
186			return Ok(v.clone());
187		}
188		let cache = if let Some(host_config) = &self.host_config {
189			let cache = nix_go!(host_config.cache);
190			let enable: bool = nix_go_json!(cache.enable);
191			if enable {
192				let kind: String = nix_go_json!(cache.kind);
193				let name: String = nix_go_json!(cache.name);
194				match kind.as_str() {
195					"attic" => {
196						let pin: bool = nix_go_json!(cache.pin);
197						Some(BinaryCache::Attic { name, pin })
198					}
199					v => bail!("unknown binary cache kind: {v}"),
200				}
201			} else {
202				None
203			}
204		} else {
205			None
206		};
207		let _ = self.binary_cache.set(cache.clone());
208		Ok(cache)
209	}
159	pub async fn deploy_kind(&self) -> Result<DeployKind> {210	pub async fn deploy_kind(&self) -> Result<DeployKind> {
160		self.deploy_kind.get_or_try_init(|| async {211		self.deploy_kind.get_or_try_init(|| async {
161			let remowt = self.remowt().await?;212			let remowt = self.remowt().await?;
238					let built = plugin289					let built = plugin
239						.build("out")290						.build("out")
240						.context("failed to build the fleet nix plugin")?;291						.context("failed to build the fleet nix plugin")?;
292					let pin = format!("{}-remowt-plugin-fleet", self.gc_root_prefix());
241					let copied = self293					let copied = self
242						.remote_derivation(&built)294						// TODO: Pin? But this pin will be host-specific, and heavy because of nix dependency...
295						.remote_derivation(&built, Some(&pin))
243						.await296						.await
244						.context("failed to copy the fleet nix plugin to the host store")?;297						.context("failed to copy the fleet nix plugin to the host store")?;
245					let bin = copied.join("bin/remowt-plugin-fleet");298					let bin = copied.join("bin/remowt-plugin-fleet");
248						.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())301						.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())
249						.await302						.await
250						.context("failed to load the fleet nix plugin")?;303						.context("failed to load the fleet nix plugin")?;
304					// 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.
305					self.remowt()
306						.await?
307						.rpc()
308						.wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))
309						.await
310						.map_err(|_| anyhow!("failed to wait"))?;
251					anyhow::Ok(())311					anyhow::Ok(())
252				})312				})
253				.await?;313				.await?;
339	}399	}
340	/// Returns path for futureproofing, as path might change i.e on conversion to CA400	/// Returns path for futureproofing, as path might change i.e on conversion to CA
341	#[instrument(skip(self))]401	#[instrument(skip(self))]
342	pub async fn remote_derivation(&self, path: &Utf8Path) -> Result<Utf8PathBuf> {402	pub async fn remote_derivation(
403		&self,
404		path: &Utf8Path,
405		pin_name: Option<&str>,
406	) -> Result<Utf8PathBuf> {
343		let path = path.to_owned();407		let path = path.to_owned();
344		if self.local {408		if self.local {
345			// Path is located locally, thus already trusted.409			// Path is located locally, thus already trusted.
346			return Ok(path);410			return Ok(path);
347		}411		}
412		let cache = self.binary_cache()?;
413		let mut cache = if let Some(cache) = cache {
414			let local = self.config.local_host().remowt().await?;
415			let path = path.to_owned();
416			let pin_name = pin_name.map(str::to_owned);
417			Some(tokio::spawn(async move {
418				if let Err(e) = cache.upload(&local, &path, pin_name.as_deref()).await {
419					warn!("failed to upload closure {path} to cache: {e:?}");
420				}
421			}))
422		} else {
423			None
424		};
348		let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {425		let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {
349			let path = path.clone();426			let path = path.clone();
350			let graph = DrvGraph::resolve(&eval_store(), &path)
351				.context("failed to resolve graph to be uploaded")?;
352			info!("signing {} paths", graph.nodes.len());
353427
354			Box::pin(async move {428			Box::pin(async move {
355				let local = self.config.local_host();429				let local = self.config.local_host();
377				.expect("copy_to panicked")451				.expect("copy_to panicked")
378				.context("copying closure to remote store")?;452				.context("copying closure to remote store")?;
379		}453		}
454
455		if let Some(cache) = cache.take() {
456			let _ = cache.await;
457		}
458
380		Ok(path)459		Ok(path)
381	}460	}
461
462	/// Like [`Self::remote_derivation`], but instead of copying the closure over
463	/// the nix daemon tunnel, uploads it to an attic binary cache and has the
464	/// remote host substitute it from there. The remote is expected to already
465	/// trust the cache's signing key and have it configured as a substituter.
466	#[instrument(skip(self))]
467	pub async fn remote_derivation_attic(
468		&self,
469		path: &Utf8Path,
470		cache_name: &str,
471	) -> Result<Utf8PathBuf> {
472		let path = path.to_owned();
473		if self.local {
474			// Path is located locally, thus already trusted.
475			return Ok(path);
476		}
477
478		info!("uploading {path} closure to attic cache {cache_name}");
479		let status = Command::new("attic")
480			.arg("push")
481			.arg(cache_name)
482			.arg(path.as_str())
483			.status()
484			.await
485			.context("failed to spawn attic-cli, is it on PATH?")?;
486		ensure!(status.success(), "attic push failed: {status}");
487
488		info!("substituting {path} on the remote host");
489		let nix = self.nix_client().await?;
490		let substituted = nix
491			.substitute(vec![path.clone()])
492			.await
493			.map_err(|e| anyhow!("{e:?}"))?
494			.map_err(|e| anyhow!("{e}"))?;
495		ensure!(
496			substituted.contains(&path),
497			"remote host failed to substitute {path} from attic cache {cache_name}"
498		);
499		Ok(path)
500	}
382}501}
383502
384struct HostSecretDefinition(Value);503struct HostSecretDefinition(Value);
385504
386impl ConfigHost {505impl ConfigHost {
506	pub fn gc_root_prefix(&self) -> String {
507		format!("{}-{}", self.config.gc_root_prefix(), self.name)
508	}
509
387	// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,510	// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,
388	// assuming getting tags always returns the same value.511	// assuming getting tags always returns the same value.
389	pub fn tags(&self) -> Result<Vec<String>> {512	pub fn tags(&self) -> Result<Vec<String>> {
504						cell627						cell
505					},628					},
506					pkgs_override: Some(self.default_pkgs.clone()),629					pkgs_override: Some(self.default_pkgs.clone()),
630					binary_cache: OnceCell::new(),
507631
508					local: true,632					local: true,
509					remowt: OnceCell::new(),633					remowt: OnceCell::new(),
546			nixos_unchecked_config: OnceLock::new(),670			nixos_unchecked_config: OnceLock::new(),
547			groups: OnceLock::new(),671			groups: OnceLock::new(),
548			pkgs_override: None,672			pkgs_override: None,
673			binary_cache: OnceCell::new(),
549674
550			// TODO: Remove with connectivit refactor675			// TODO: Remove with connectivit refactor
551			local: self.localhost == name,676			local: self.localhost == name,

9use crate::{fleetdata::SecretOwner, host::Config};9use crate::{fleetdata::SecretOwner, host::Config};
1010
11impl Config {11impl Config {
12	pub fn gc_root_prefix(&self) -> String {
13		self.data.gc_root_prefix.clone()
14	}
12	fn cached_host_key(&self, host: &str) -> Option<String> {15	fn cached_host_key(&self, host: &str) -> Option<String> {
13		let hosts = self.data.hosts.read().expect("no poisoning");16		let hosts = self.data.hosts.read().expect("no poisoning");
14		let key = hosts.get(host).map(|h| &h.encryption_key);17		let key = hosts.get(host).map(|h| &h.encryption_key);

119				.context("failed to build generator for target host")?;119				.context("failed to build generator for target host")?;
120120
121			let generator = host_on121			let generator = host_on
122				.remote_derivation(&generator)122				.remote_derivation(&generator, None)
123				.await123				.await
124				.context("failed to copy generator to target host")?;124				.context("failed to copy generator to target host")?;
125125

39	make_attrs, make_bindings_builder, make_list, make_list_builder, realised_string,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,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,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,42	set_err_msg, setting_get, 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,43	store_parse_path, store_path_free, store_path_name, string_realise, value, value_call,
44	value_decref, value_force, value_incref,44	value_decref, value_force, value_incref,
45};45};
390	with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())390	with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())
391}391}
392
393pub fn get_setting(s: &CStr) -> Result<String> {
394	let mut out = String::new();
395	with_default_context(|c, _| unsafe {
396		setting_get(c, s.as_ptr(), Some(copy_nix_str), (&raw mut out).cast())
397	})?;
398	Ok(out)
399}
392400
393#[derive(Debug)]401#[derive(Debug)]
394pub struct AddedFile {402pub struct AddedFile {

109			(ActivityType::CopyPaths, []) => {109			(ActivityType::CopyPaths, []) => {
110				debug_span!(target: "nix::copy-paths", "copying paths")110				debug_span!(target: "nix::copy-paths", "copying paths")
111			}111			}
112			// Umbrella progress activity. Nix emits it at lvlError ("always show"),
113			// which would otherwise become a noisy ERROR `action` span — give it an
114			// explicit debug level like the other container activities.
115			(ActivityType::Builds, []) => {
116				debug_span!(target: "nix::builds", "building")
117			}
112			(ActivityType::Unknown, [])118			(ActivityType::Unknown, [])
113				if s.starts_with("copying \"") && s.ends_with("\" to the store") =>119				if s.starts_with("copying \"") && s.ends_with("\" to the store") =>
114			{120			{
365	BuildGraphGuard { paths }371	BuildGraphGuard { paths }
366}372}
373
374/// Drop the lazily-created `building` span for a derivation. Used when a build
375/// fails: its dependents will never build, so their speculatively-created spans
376/// (made when a dependency started building) would otherwise linger until the
377/// whole build graph is torn down.
378pub fn remove_drv_span(drv_path: &Utf8Path) {
379	if let Some(entry) = DRV_GRAPH.lock().expect("not poisoned").get_mut(drv_path) {
380		entry.span = None;
381	}
382}
367383
368fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {384fn ensure_drv_span(drv_path: &Utf8Path) -> Option<Span> {
369	let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");385	let mut drv_graph = DRV_GRAPH.lock().expect("not poisoned");
524			// ResultType::FileLinked => todo!(),540			// ResultType::FileLinked => todo!(),
525			(ResultType::BuildLogLine, [Str(s)]) => {541			(ResultType::BuildLogLine, [Str(s)]) => {
526				let s = ansi_filter(s);542				let s = ansi_filter(s);
527				info!("{s}");543				info!(target: "nix::build", "{s}");
528			}544			}
529			// ResultType::UntrustedPath => todo!(),545			// ResultType::UntrustedPath => todo!(),
530			// ResultType::CorruptedPath => todo!(),546			// ResultType::CorruptedPath => todo!(),

45	},45	},
46}46}
47
48// Derivations whose names contain any of these are "heavy": memory-hungry builds that already
49// saturate all cores internally (e.g. composable-kernels), so running anything alongside them
50// just thrashes RAM. A heavy build grabs every semaphore permit and runs exclusively.
51const DEFAULT_HEAVY_DRVS: &[&str] = &[
52	"composable-kernel",
53	"composable_kernel",
54	"ghc",
55	"gfortran",
56	"llvm",
57	"clang",
58];
59
60fn heavy_patterns_from_env() -> Vec<String> {
61	let mut pats: Vec<String> = DEFAULT_HEAVY_DRVS.iter().map(|s| (*s).to_owned()).collect();
62	if let Ok(extra) = std::env::var("FLEET_HEAVY_DRVS") {
63		pats.extend(
64			extra
65				.split(',')
66				.map(str::trim)
67				.filter(|s| !s.is_empty())
68				.map(str::to_owned),
69		);
70	}
71	pats
72}
4773
48pub struct Scheduler {74pub struct Scheduler {
49	store: Arc<Store>,75	store: Arc<Store>,
50	parallelism: usize,76	parallelism: usize,
77	heavy_patterns: Vec<String>,
51	events: broadcast::Sender<BuildEvent>,78	events: broadcast::Sender<BuildEvent>,
52}79}
5380
58		Self {85		Self {
59			store: eval_store(),86			store: eval_store(),
60			parallelism,87			parallelism,
88			heavy_patterns: heavy_patterns_from_env(),
61			events,89			events,
62		}90		}
63	}91	}
92
93	// Permits a drv must hold to run. Heavy drvs take all of them, pausing everything else.
94	fn permits_for(&self, name: &str) -> u32 {
95		if self.heavy_patterns.iter().any(|p| name.contains(p)) {
96			self.parallelism as u32
97		} else {
98			1
99		}
100	}
64101
65	pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {102	pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {
66		self.events.subscribe()103		self.events.subscribe()
139						.get(&path)176						.get(&path)
140						.map(|n| n.name.clone())177						.map(|n| n.name.clone())
141						.unwrap_or_default();178						.unwrap_or_default();
179					crate::logging::remove_drv_span(&path);
142					let _ = self.events.send(BuildEvent::DrvCancelled {180					let _ = self.events.send(BuildEvent::DrvCancelled {
143						drv_path: path.clone(),181						drv_path: path.clone(),
144						name,182						name,
153				let graph = graph.clone();191				let graph = graph.clone();
154				let wanted_here = wanted.get(&path).cloned().unwrap_or_default();192				let wanted_here = wanted.get(&path).cloned().unwrap_or_default();
155				let store = self.store.clone();193				let store = self.store.clone();
194				let permits = graph
195					.nodes
196					.get(&path)
197					.map(|n| self.permits_for(&n.name))
198					.unwrap_or(1);
156				in_flight.push(tokio::spawn(async move {199				in_flight.push(tokio::spawn(async move {
200					if permits > 1 {
201						debug!(permits, "heavy drv: waiting for exclusive build slot");
202					}
203					// Heavy drvs acquire every permit, so they only start once all in-flight
204					// builds drain and nothing new can slip in (the semaphore is FIFO-fair).
157					let _permit = sem.acquire_owned().await.expect("semaphore not closed");205					let _permit = sem
206						.acquire_many_owned(permits)
207						.await
208						.expect("semaphore not closed");
158					let node = graph209					let node = graph
224					propagate_done(&dependents, &mut indeg, &mut ready, &finished);275					propagate_done(&dependents, &mut indeg, &mut ready, &finished);
225				}276				}
226				Err(e) => {277				Err(e) => {
278					crate::logging::remove_drv_span(&finished);
227					failed.insert(finished.clone(), format!("{e:#}"));279					failed.insert(finished.clone(), format!("{e:#}"));
228					mark_tainted(&dependents, &finished, &mut tainted);280					mark_tainted(&dependents, &finished, &mut tainted);
229					propagate_done(&dependents, &mut indeg, &mut ready, &finished);281					propagate_done(&dependents, &mut indeg, &mut ready, &finished);
368// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing420// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing
369// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one421// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one
370// will work as a hard cap.422// will work as a hard cap.
423/// Number of derivations to build concurrently, taken from Nix's `max-jobs`
424/// setting. `auto` resolves to the CPU count, matching Nix itself.
425fn max_jobs() -> usize {
426	let cores = || {
427		std::thread::available_parallelism()
428			.map(|p| p.get())
429			.unwrap_or(4)
430	};
431	match crate::get_setting(c"max-jobs") {
432		Ok(v) if v.trim() == "auto" => cores(),
433		Ok(v) => v.trim().parse().unwrap_or_else(|_| cores()),
434		Err(_) => cores(),
435	}
436}
437
371pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {438pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {
372	let parallelism = std::thread::available_parallelism()
373		.map(|p| p.get())
374		.unwrap_or(4);
375	let scheduler = Scheduler::new(parallelism);439	let scheduler = Scheduler::new(max_jobs());
376	crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })440	crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })
377		.context("scheduler run")441		.context("scheduler run")
378}442}

28	Sign(String),28	Sign(String),
29	#[error("listing generations failed: {0}")]29	#[error("listing generations failed: {0}")]
30	ListGenerations(String),30	ListGenerations(String),
31	#[error("substitution failed: {0}")]
32	Substitute(String),
31}33}
3234
33#[endpoints(ns = 91)]35#[endpoints(ns = 91)]
60		.map_err(|e| NixError::Sign(e.to_string()))62		.map_err(|e| NixError::Sign(e.to_string()))
61	}63	}
64
65	/// Download the given paths (and their closures) into the local store using
66	/// the configured substituters. Used to fetch closures that were uploaded to
67	/// a binary cache (e.g. attic) instead of copied over the nix daemon tunnel.
68	#[endpoints(id = 6)]
69	async fn substitute(&self, paths: Vec<Utf8PathBuf>) -> Result<Vec<Utf8PathBuf>, NixError> {
70		spawn_blocking(move || {
71			let store = eval_store();
72			store.substitute_paths(&paths)
73		})
74		.await
75		.expect("substitution panicked")
76		.map_err(|e| NixError::Substitute(e.to_string()))
77	}
6278
63	#[endpoints(id = 5)]79	#[endpoints(id = 5)]
64	async fn list_generations(80	async fn list_generations(