Something went wrong. Try again.
Nix-based distributed command runner now.dev.br
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554// now: Nix-based distributed command runner// Copyright (C) 2026 Eric Rodrigues Pires//// This program is free software: you can redistribute it and/or modify it under// the terms of the GNU Affero General Public License as published by the Free// Software Foundation, either version 3 of the License, or (at your option)// any later version.//// This program is distributed in the hope that it will be useful, but WITHOUT// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS// FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for// more details.//// You should have received a copy of the GNU Affero General Public License along// with this program. If not, see <https://www.gnu.org/licenses/>.
use std::{ env::temp_dir, ffi::{OsStr, OsString}, num::NonZeroUsize, os::unix::ffi::OsStrExt, path::{Path, PathBuf}, time::Duration,};
use ahash::{HashMap, HashSet};use async_trait::async_trait;use color_eyre::eyre::{OptionExt, eyre};use futures::{AsyncWriteExt, stream::FuturesUnordered};use smol::{ channel::{self, Receiver}, lock::{ RwLock, RwLockReadGuard, RwLockWriteGuard, Semaphore, SemaphoreGuard, futures::{Acquire, Read, Write}, }, process::{Child, Command, Stdio}, stream::StreamExt,};
use crate::{ builder::{NixConfig, NowBuilder, remote::RemoteBuilder}, environment::NowEnvironment, job::JobError, utils::{wait_for_output, write_output_to_stderr}, workflow::{NowCheckout, NowJob, NowRunMode},};
pub(crate) struct LocalBuilder { pub(crate) cancellation: channel::Sender<()>, pub(crate) receiver: channel::Receiver<()>, pub(crate) semaphore: Semaphore, pub(crate) lock: RwLock<()>, pub(crate) env: HashMap<OsString, OsString>, pub(crate) hostname: String, pub(crate) extra_platforms: Vec<String>, pub(crate) system: String, pub(crate) system_features: HashSet<String>, pub(crate) remote_builders: Vec<RemoteBuilder>, pub(crate) run_mode: NowRunMode, pub(crate) gcroot_dir: PathBuf,}
pub(crate) struct BuilderGuard<'a> { pub(crate) guard: SemaphoreGuard<'a>, pub(crate) lock: RwLockReadGuard<'a, ()>, pub(crate) receiver: &'a Receiver<()>, pub(crate) builder: &'a dyn NowBuilder,}
pub(crate) struct RunnerGuard<'a> { pub(crate) lock: RwLockWriteGuard<'a, ()>, pub(crate) receiver: &'a Receiver<()>, pub(crate) builder: &'a dyn NowBuilder,}
impl LocalBuilder { pub(crate) async fn new( environment: &NowEnvironment, builders: Option<String>, run_mode: NowRunMode, cores: Option<NonZeroUsize>, keep_alive: Option<Duration>, ) -> color_eyre::Result<Self> { let mut command = Command::new("nix"); command .args([ "--extra-experimental-features", "nix-command flakes", "config", "show", "--json", ]) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped());
let mut child = command.spawn()?; let output = wait_for_output(&mut child, None).await?; let config: NixConfig = if output.status.success() { serde_json::from_slice(&output.stdout)? } else { write_output_to_stderr(&output)?; return Err(eyre!("Failed to fetch Nix config")); };
let remote_builders = match run_mode { NowRunMode::All => { RemoteBuilder::get_remote_builders( &config, builders, environment.nix_project_source.as_ref(), keep_alive, ) .await? } NowRunMode::LocalOnly => vec![], NowRunMode::RemoteOnly => { let remote_builders = RemoteBuilder::get_remote_builders( &config, builders, environment.nix_project_source.as_ref(), keep_alive, ) .await?; if remote_builders.is_empty() { return Err(eyre!("No remote builders available")); } remote_builders } };
let (cancellation, receiver) = channel::bounded(1);
let cores = match cores { Some(cores) => cores.into(), None => smol::unblock(sysinfo::System::physical_core_count) .await .unwrap_or(1), };
let hostname = smol::unblock(sysinfo::System::host_name) .await .unwrap_or_else(|| "localhost".into());
let mut env = environment.local_env.clone(); env.insert( "NOW_GCROOT_DIR".into(), environment.gcroot_dir.clone().into(), );
Ok(Self { cancellation, receiver, semaphore: Semaphore::new(cores), lock: RwLock::default(), env, hostname, extra_platforms: config.extra_platforms.value, system: config.system.value, system_features: config.system_features.value.into_iter().collect(), remote_builders, run_mode, gcroot_dir: environment.gcroot_dir.clone(), }) }
pub(crate) fn cancel_builders(&self) { self.cancellation.close(); for remote in &self.remote_builders { remote.cancellation.close(); } }
pub(crate) async fn get_builder( &self, job: &NowJob, ) -> color_eyre::Result<Option<BuilderGuard<'_>>> { let mut builders = vec![];
match self.run_mode { NowRunMode::All | NowRunMode::LocalOnly => { if job.build_system == self.system && job .required_system_features .iter() .all(|feature| self.system_features.contains(feature)) { builders.push(self as &dyn NowBuilder); } } NowRunMode::RemoteOnly => {} }
for builder in &self.remote_builders { if builder.build_systems.contains(&job.build_system) && builder .required_features .iter() .all(|feature| job.required_system_features.contains(feature)) && job .required_system_features .iter() .all(|feature| builder.system_features.contains(feature)) { builders.push(builder as &dyn NowBuilder); } }
let mut builders_fut: FuturesUnordered<_> = builders .into_iter() .map(|builder| async move { let (semaphore, read, receiver) = builder.acquire(); let lock = read.await; let guard = semaphore.await; BuilderGuard { guard, lock, receiver, builder, } }) .collect();
Ok(builders_fut.next().await) }
pub(crate) fn has_runner(&self, job: &NowJob) -> bool { match self.run_mode { NowRunMode::All | NowRunMode::LocalOnly => { if (job.host_system == self.system || self.extra_platforms.contains(&job.host_system)) && job .required_system_features .iter() .all(|feature| self.system_features.contains(feature)) { return true; } } NowRunMode::RemoteOnly => {} }
for builder in &self.remote_builders { if builder.host_system == job.host_system && builder .required_features .iter() .all(|feature| job.required_system_features.contains(feature)) && job .required_system_features .iter() .all(|feature| builder.system_features.contains(feature)) { return true; } }
false }
pub(crate) async fn get_runner( &self, job: &NowJob, ) -> color_eyre::Result<Option<RunnerGuard<'_>>> { let mut runners = vec![];
match self.run_mode { NowRunMode::All | NowRunMode::LocalOnly => { if (job.host_system == self.system || self.extra_platforms.contains(&job.host_system)) && job .required_system_features .iter() .all(|feature| self.system_features.contains(feature)) { runners.push(self as &dyn NowBuilder); } } NowRunMode::RemoteOnly => {} }
for builder in &self.remote_builders { if builder.host_system == job.host_system && builder .required_features .iter() .all(|feature| job.required_system_features.contains(feature)) && job .required_system_features .iter() .all(|feature| builder.system_features.contains(feature)) { runners.push(builder as &dyn NowBuilder); } }
let mut runners_fut: FuturesUnordered<_> = runners .into_iter() .map(|builder| async move { let (write, receiver) = builder.acquire_exclusive(); let lock = write.await; RunnerGuard { lock, receiver, builder, } }) .collect();
Ok(runners_fut.next().await) }}
#[async_trait(?Send)]impl NowBuilder for LocalBuilder { fn acquire(&self) -> (Acquire<'_>, Read<'_, ()>, &channel::Receiver<()>) { (self.semaphore.acquire(), self.lock.read(), &self.receiver) }
fn acquire_exclusive(&self) -> (Write<'_, ()>, &channel::Receiver<()>) { (self.lock.write(), &self.receiver) }
fn get_name(&self) -> String { self.hostname.clone() }
fn is_remote(&self) -> bool { false }
async fn create_tempdir( &self, checkout: NowCheckout, cancellation: &channel::Receiver<()>, ) -> Result<Option<PathBuf>, JobError> { match checkout { NowCheckout::Default | NowCheckout::All => Ok(None), NowCheckout::None | NowCheckout::Clone | NowCheckout::CloneAll => { let mut tempdir_command = Command::new("sh"); tempdir_command .args(["-c", r#"mktemp -d "${TMPDIR:-/tmp}/now-XXXXXXXXXX""#]) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped());
let mut child = tempdir_command.spawn().map_err(|error| eyre!(error))?; let output = wait_for_output(&mut child, Some(cancellation)).await?; if output.status.success() { Ok(Some(PathBuf::from(OsStr::from_bytes( output.stdout.trim_ascii(), )))) } else { write_output_to_stderr(&output)?; Err(eyre!("Temporary directory creation failed").into()) } } } }
async fn checkout( &self, checkout: NowCheckout, tempdir: Option<PathBuf>, cancellation: &channel::Receiver<()>, ) -> Result<PathBuf, JobError> { match checkout { NowCheckout::Default | NowCheckout::All => { debug_assert!(tempdir.is_none()); Ok(std::env::current_dir().map_err(|error| eyre!(error))?) } NowCheckout::None => { let tempdir = tempdir.ok_or_eyre(eyre!("Missing temporary directory"))?; Ok(tempdir) } NowCheckout::Clone => { let tempdir = tempdir.ok_or_eyre(eyre!("Missing temporary directory"))?; let mut command = Command::new("rsync"); command .args(["-arz", "--files-from=-", "."]) .arg(&tempdir) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped());
let mut child = command.spawn().map_err(|error| eyre!(error))?; let mut stdin = child.stdin.take().expect("stdin is piped"); let stdin_future = async move { let cwd = std::env::current_dir().map_err(|error| eyre!(error))?; for dir_entry in ignore::Walk::new(std::env::current_dir().map_err(|error| eyre!(error))?) .flatten() { if dir_entry.file_type().is_some_and(|typ| typ.is_file()) { let path = dir_entry.path(); stdin .write_all( path.strip_prefix(&cwd) .unwrap_or(path) .as_os_str() .as_bytes(), ) .await .map_err(|error| eyre!(error))?; stdin.write_all(b"\n").await.map_err(|error| eyre!(error))?; } } Ok(stdin.flush().await.map_err(|error| eyre!(error))?) }; let (_, output) = smol::future::try_zip( stdin_future, wait_for_output(&mut child, Some(cancellation)), ) .await?; if output.status.success() { Ok(tempdir) } else { write_output_to_stderr(&output)?; Err(eyre!("Checkout failed").into()) } } NowCheckout::CloneAll => { let tempdir = tempdir.ok_or_eyre(eyre!("Missing temporary directory"))?;
let mut command = Command::new("rsync"); command .args(["-arz", "."]) .arg(&tempdir) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped());
let mut child = command.spawn().map_err(|error| eyre!(error))?; let output = wait_for_output(&mut child, Some(cancellation)).await?; if output.status.success() { Ok(tempdir) } else { write_output_to_stderr(&output)?; Err(eyre!("Checkout failed").into()) } } } }
async fn copy_derivations( &self, _job_name: &str, derivations: &[PathBuf], _cancellation: &channel::Receiver<()>, ) -> Result<(), JobError> { debug_assert!(derivations.iter().all(|path| path.exists())); Ok(()) }
async fn realize_derivation( &self, derivation: &Path, cancellation: &channel::Receiver<()>, ) -> Result<PathBuf, JobError> { let gcroot = self .gcroot_dir .join(derivation.file_name().expect("derivation has file name"));
let mut command = Command::new("nix-store"); command .arg("--add-root") .arg(&gcroot) .args(["--indirect", "--realise"]) .arg(derivation) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped());
let mut child = command.spawn().map_err(|error| eyre!(error))?; let output = wait_for_output(&mut child, Some(cancellation)).await?; if output.status.success() { Ok(PathBuf::from(OsStr::from_bytes(output.stdout.trim_ascii()))) } else { write_output_to_stderr(&output)?; Err(eyre!( "Failed to realize derivation '{}'", derivation.to_string_lossy(), ) .into()) } }
async fn download( &self, downloads: &[PathBuf], _cancellation: &channel::Receiver<()>, ) -> Result<(), JobError> { debug_assert!(downloads.iter().all(|path| path.exists())); Ok(()) }
fn run_derivation(&self, cwdir: &Path, derivation: PathBuf) -> Result<Child, JobError> { let mut command = Command::new(derivation.join("bin/now-step")); command .current_dir(cwdir) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .env_clear() .envs(&self.env); Ok(command.spawn().map_err(|error| eyre!(error))?) }
async fn fetch_derivation( &self, derivation: &Path, _cancellation: &channel::Receiver<()>, ) -> Result<(), JobError> { debug_assert!(derivation.exists()); Ok(()) }
async fn undo_checkout(&self, checkout: NowCheckout, path: &Path) -> Result<(), JobError> { match checkout { NowCheckout::Default | NowCheckout::All => { debug_assert_eq!( path.canonicalize().map_err(|error| eyre!(error))?, std::env::current_dir() .map_err(|error| eyre!(error))? .canonicalize() .map_err(|error| eyre!(error))? ); Ok(()) } NowCheckout::None | NowCheckout::Clone | NowCheckout::CloneAll => { assert_ne!( path.canonicalize().map_err(|error| eyre!(error))?, std::env::current_dir() .map_err(|error| eyre!(error))? .canonicalize() .map_err(|error| eyre!(error))? ); assert!(path.starts_with(temp_dir())); let mut command = Command::new("rm"); command.arg("-rf").arg(path);
let mut child = command.spawn().map_err(|error| eyre!(error))?; let output = wait_for_output(&mut child, None).await?; if output.status.success() { Ok(()) } else { write_output_to_stderr(&output)?; Err(eyre!("Failed to remove locally checked out directory").into()) } } } }}