From 6386febdb81eb1e072ad7cb568019f06618008fe Mon Sep 17 00:00:00 2001 From: Tom Alexander Date: Fri, 31 Jul 2026 19:26:46 -0400 Subject: [PATCH 1/2] Verify the nix store before starting a build run. --- src/command/build/runner.rs | 9 ++ src/nix_util/high_level.rs | 24 +++++ src/nix_util/mod.rs | 2 + src/nix_util/running_verify.rs | 181 +++++++++++++++++++++++++++++++++ 4 files changed, 216 insertions(+) create mode 100644 src/nix_util/running_verify.rs diff --git a/src/command/build/runner.rs b/src/command/build/runner.rs index 5517691..3ff03f7 100644 --- a/src/command/build/runner.rs +++ b/src/command/build/runner.rs @@ -12,6 +12,7 @@ use crate::git_util::git_force_into_state; use crate::git_util::git_init_at_rev; use crate::nix_util::FlakeLock; use crate::nix_util::nix_flake_update; +use crate::nix_util::nix_store_verify_repair_check_contents; use crate::nix_util::nixos_build_target; use crate::nix_util::parse_flake_lock; @@ -33,6 +34,8 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> { let db_handle = DbHandle::new(Some(database_path)).await?; + verify_nix_store().await?; + for target_name in args.target { let target_config = { let target_config = config.get_target_config(&target_name)?; @@ -85,6 +88,12 @@ async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig) Ok(()) } +async fn verify_nix_store() -> Result<()> { + nix_store_verify_repair_check_contents().await?; + + Ok(()) +} + async fn run_nix_update(config_root: &Config, target_config: &TargetConfig) -> Result<()> { let flake_directory = target_config.get_flake_directory(config_root)?; diff --git a/src/nix_util/high_level.rs b/src/nix_util/high_level.rs index 82653f6..9c26faa 100644 --- a/src/nix_util/high_level.rs +++ b/src/nix_util/high_level.rs @@ -9,6 +9,7 @@ use crate::Result; use crate::database::db_handle::DbHandle; use super::RunningUpdate; +use super::RunningVerify; use super::running_build::RunningBuild; pub(crate) async fn nixos_build_target( @@ -87,3 +88,26 @@ where Ok(()) } + +pub(crate) async fn nix_store_verify_repair_check_contents() -> Result<()> { + let mut command = Command::new("nix-store"); + command.stdout(Stdio::piped()); + command.stderr(Stdio::piped()); + command.stdin(Stdio::null()); + command.args([ + "--verify", + "--check-contents", + "--repair", + "--log-format", + "internal-json", + "-vvvvvvvvvvv", + ]); + command.kill_on_drop(true); + + let child = command.spawn()?; + + let mut running_verify = RunningVerify::new()?; + running_verify.run_to_completion(child).await?; + + Ok(()) +} diff --git a/src/nix_util/mod.rs b/src/nix_util/mod.rs index 603e8cb..f910078 100644 --- a/src/nix_util/mod.rs +++ b/src/nix_util/mod.rs @@ -7,6 +7,7 @@ mod nix_output_stream; mod output_stream; mod running_build; mod running_update; +mod running_verify; mod tree_iter; pub(crate) use activity_tree::ActivityIdAlreadyInTreeError; pub(crate) use activity_tree::ActivityIdNotInTreeError; @@ -18,3 +19,4 @@ pub(crate) use output_stream::OutputLine; pub(crate) use output_stream::OutputLineStream; pub(crate) use running_build::RunningBuild; pub(crate) use running_update::RunningUpdate; +pub(crate) use running_verify::RunningVerify; diff --git a/src/nix_util/running_verify.rs b/src/nix_util/running_verify.rs new file mode 100644 index 0000000..d660fcb --- /dev/null +++ b/src/nix_util/running_verify.rs @@ -0,0 +1,181 @@ +use std::borrow::Cow; +use std::time::Duration; +use std::time::Instant; + +use tokio::process::Child; +use tracing::error; +use tracing::info; + +use crate::Result; +use crate::nix_util::NixOutputStream; +use crate::nix_util::nix_output_stream::NixAction; +use crate::nix_util::output_stream::OutputStream; +use crate::nix_util::tree_iter::ForwardTreeIter; + +use super::activity_tree::ActivityTreeEntry; +use super::activity_tree_stream::ActivityTreeStream; +use super::nix_output_stream::ActivityResultMessage; +use super::nix_output_stream::NixMessage; + +pub(crate) struct RunningVerify { + activity_tree: ActivityTreeStream, + last_announce: Option, +} + +impl RunningVerify { + pub(crate) fn new() -> Result { + Ok(RunningVerify { + activity_tree: ActivityTreeStream::new(), + last_announce: None, + }) + } + + pub(crate) async fn run_to_completion(&mut self, mut child: Child) -> Result<()> { + let output_stream = OutputStream::from_child(&mut child)?; + let mut nix_output_stream: NixOutputStream = + NixOutputStream::new(output_stream); + + info!("Verifying nix store."); + + let exit_status_handle = tokio::spawn(async move { + let status = child + .wait() + .await + .expect("nixos-rebuild encountered an error"); + status + }); + + while let Some(message) = nix_output_stream.next().await? { + self.handle_message(message)?; + } + + let exit_status = exit_status_handle.await?; + info!("nix store verify status was: {}", exit_status); + + Ok(()) + } + + pub(crate) fn handle_message(&mut self, message: NixMessage) -> Result<()> { + self.activity_tree.handle_message(&message)?; + + let message = match message { + NixMessage::ParseFailure(line) => { + error!("FAIL PARSE: {line}"); + return Ok(()); + } + NixMessage::Generic(_value, line) => { + error!("GENERIC PARSE: {line}"); + return Ok(()); + } + NixMessage::Action(nix_action) => nix_action, + }; + match message { + NixAction::Msg(msg_message) => { + // if msg_message.level > 0 && msg_message.level < 5 { + // eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg); + // } + } + NixAction::Start(activity_start_message) => { + // println!("START: {}", serde_json::to_string(&activity_start_message)?); + self.print_current_status(); + } + NixAction::Stop(stop_message) => { + // println!("STOP: {}", serde_json::to_string(&stop_message)?); + self.print_current_status(); + } + NixAction::Result(activity_result_message) => { + match activity_result_message { + ActivityResultMessage::FileLinked(_activity_result_file_linked) => {} + ActivityResultMessage::BuildLogLine(_activity_result_build_log_line) => {} + ActivityResultMessage::UntrustedPath(_activity_result_untrusted_path) => {} + ActivityResultMessage::CorruptedPath(_activity_result_corrupted_path) => {} + ActivityResultMessage::SetPhase(_activity_result_set_phase) => {} + ActivityResultMessage::Progress(activity_result_progress) => { + // if activity_result_progress.expected != 0 { + // println!( + // "PROGRESS: {}", + // serde_json::to_string(&activity_result_progress)? + // ); + // } + self.maybe_print_current_status(); + } + ActivityResultMessage::SetExpected(activity_result_set_expected) => { + // if activity_result_set_expected.expected != 0 { + // println!( + // "EXPECTED: {}", + // serde_json::to_string(&activity_result_set_expected)? + // ); + // } + self.maybe_print_current_status(); + } + ActivityResultMessage::PostBuildLogLine( + _activity_result_post_build_log_line, + ) => {} + ActivityResultMessage::FetchStatus(_activity_result_fetch_status) => {} + }; + } + }; + Ok(()) + } + + fn maybe_print_current_status(&mut self) -> () { + let last_announce = match self.last_announce { + Some(instant) => instant, + None => { + // If we haven't announced before, always announce. + return self.print_current_status(); + } + }; + + let now = Instant::now(); + let time_since_last_announce = now.duration_since(last_announce); + if time_since_last_announce > Duration::new(5, 0) { + return self.print_current_status(); + } + } + + fn print_current_status(&mut self) -> () { + let nodes = ForwardTreeIter::new( + self.activity_tree.get_tree(), + None, + is_match_predicate, + is_transparent_predicate, + is_alive_predicate, + ); + let mut out = Vec::new(); + for (depth, _is_match, _is_transparent, node) in nodes { + let progress_text = node + .get_activity() + .get_progress_text() + .unwrap_or(Cow::Borrowed("")); + let name = node + .get_activity() + .display_name() + .unwrap_or(Cow::Borrowed("null")); + out.push(format!("{depth}\t{progress_text}\t{name}")) + } + + if out.is_empty() { + println!("No active activities."); + } else { + println!("\n"); + for l in out { + println!("{l}\n"); + } + println!("\n"); + } + self.last_announce = Some(Instant::now()); + } +} + +fn is_match_predicate(entry: &ActivityTreeEntry) -> bool { + entry.get_activity().get_progress_text().is_some() +} + +pub(crate) fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool { + return false; +} + +fn is_alive_predicate(entry: &ActivityTreeEntry) -> bool { + entry.get_activity().is_active() +} From 0a270dcbdd51baa2895634a3baee29373db8337a Mon Sep 17 00:00:00 2001 From: Tom Alexander Date: Sat, 8 Aug 2026 23:50:23 -0400 Subject: [PATCH 2/2] Record error when setting up repo fails rather than abort entire script. --- example_config.toml | 7 +++ src/command/build/runner.rs | 91 +++++++++++++++++++++++++--------- src/nix_util/high_level.rs | 7 +-- src/nix_util/running_build.rs | 18 ++----- src/nix_util/running_verify.rs | 6 +-- 5 files changed, 85 insertions(+), 44 deletions(-) diff --git a/example_config.toml b/example_config.toml index e979de7..29f2223 100644 --- a/example_config.toml +++ b/example_config.toml @@ -1,5 +1,12 @@ # repo_directory = "/home/nixworker/persist/nix_builder" +[[targets]] + name = "foo" + repo = "https://code.fizz.buzz/talexander/machine_setup.git" + branch = "foo" + path = "nix/configuration" + attr = "nixosConfigurations.odo.config.system.build.toplevel" + [[targets]] name = "odo" repo = "https://code.fizz.buzz/talexander/machine_setup.git" diff --git a/src/command/build/runner.rs b/src/command/build/runner.rs index 3ff03f7..c13eb7d 100644 --- a/src/command/build/runner.rs +++ b/src/command/build/runner.rs @@ -1,4 +1,7 @@ +use std::process::ExitStatus; + use sqlx::Row; +use tracing::error; use tracing::info; use crate::Result; @@ -37,20 +40,46 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> { verify_nix_store().await?; for target_name in args.target { - let target_config = { - let target_config = config.get_target_config(&target_name)?; - if let Some(conf) = target_config { - conf - } else { - return Err(format!("Could not find target {}", target_name).into()); + // Record start of build + let build_id: i64 = sqlx::query( + r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#, + ) + .bind(&target_name) + .fetch_one(&db_handle.conn) + .await? + .try_get("id")?; + + // Put the rest into a function + let result = full_build_target(&config, &db_handle, &target_name, build_id).await; + match result { + Ok(exit_status) => { + let update: u64 = sqlx::query( + r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#, + ) + .bind( + exit_status + .code() + .expect("Process should have an exit code."), + ) + .bind(build_id) + .execute(&db_handle.conn) + .await? + .rows_affected(); + assert!(update == 1); + } + Err(e) => { + error!("Error building target {}: {}", target_name, e); + let update: u64 = sqlx::query( + r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#, + ) + .bind(-1) + .bind(build_id) + .execute(&db_handle.conn) + .await? + .rows_affected(); + assert!(update == 1); } }; - - prepare_flake_repo(&config, target_config).await?; - if target_config.get_update() { - run_nix_update(&config, target_config).await?; - } - build_target(&db_handle, &config, target_config).await?; } db_handle.conn.close().await; @@ -58,6 +87,29 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> { Ok(()) } +async fn full_build_target( + config: &Config, + db_handle: &DbHandle, + target_name: &str, + build_id: i64, +) -> Result { + let target_config = { + let target_config = config.get_target_config(&target_name)?; + if let Some(conf) = target_config { + conf + } else { + return Err(format!("Could not find target {}", target_name).into()); + } + }; + + prepare_flake_repo(&config, target_config).await?; + if target_config.get_update() { + run_nix_update(&config, target_config).await?; + } + let exit_status = build_target(&db_handle, &config, target_config, build_id).await?; + Ok(exit_status) +} + async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig) -> Result<()> { let repo_directory = target_config.get_repo_directory(config_root)?; assert_directory!( @@ -106,7 +158,8 @@ async fn build_target( db_handle: &DbHandle, config_root: &Config, target_config: &TargetConfig, -) -> Result<()> { + build_id: i64, +) -> Result { let flake_directory = target_config.get_flake_directory(config_root)?; let build_directory = target_config.get_build_directory(config_root)?; assert_directory!( @@ -117,18 +170,10 @@ async fn build_target( let target_name = target_config.get_name()?; - let build_id: i64 = sqlx::query( - r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#, - ) - .bind(target_name) - .fetch_one(&db_handle.conn) - .await? - .try_get("id")?; - let flake_lock = parse_flake_lock(&flake_directory).await?; write_input_revs_to_db(db_handle, &flake_lock, build_id).await?; - nixos_build_target( + let exit_status = nixos_build_target( db_handle, build_directory, flake_directory, @@ -138,7 +183,7 @@ async fn build_target( ) .await?; - Ok(()) + Ok(exit_status) } async fn write_input_revs_to_db( diff --git a/src/nix_util/high_level.rs b/src/nix_util/high_level.rs index 9c26faa..0691793 100644 --- a/src/nix_util/high_level.rs +++ b/src/nix_util/high_level.rs @@ -1,6 +1,7 @@ use std::ffi::OsStr; use std::ffi::OsString; use std::path::Path; +use std::process::ExitStatus; use std::process::Stdio; use tokio::process::Command; @@ -19,7 +20,7 @@ pub(crate) async fn nixos_build_target( attr: A, target_name: TN, build_id: i64, -) -> Result<()> +) -> Result where B: AsRef, F: AsRef, @@ -58,9 +59,9 @@ where let child = command.spawn()?; let mut running_build = RunningBuild::new(db_handle, build_id)?; - running_build.run_to_completion(child, target_name).await?; + let exit_status = running_build.run_to_completion(child, target_name).await?; - Ok(()) + Ok(exit_status) } pub(crate) async fn nix_flake_update(flake_path: F) -> Result<()> diff --git a/src/nix_util/running_build.rs b/src/nix_util/running_build.rs index f5b84ef..e411947 100644 --- a/src/nix_util/running_build.rs +++ b/src/nix_util/running_build.rs @@ -1,4 +1,5 @@ use std::borrow::Cow; +use std::process::ExitStatus; use std::time::Duration; use std::time::Instant; use std::time::SystemTime; @@ -43,7 +44,7 @@ impl<'db> RunningBuild<'db> { &mut self, mut child: Child, target_name: TN, - ) -> Result<()> + ) -> Result where TN: AsRef, { @@ -66,20 +67,7 @@ impl<'db> RunningBuild<'db> { let exit_status = exit_status_handle.await?; println!("nix build status was: {}", exit_status); - let update: u64 = - sqlx::query(r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#) - .bind( - exit_status - .code() - .expect("Process should have an exit code."), - ) - .bind(self.build_id) - .execute(&self.db_handle.conn) - .await? - .rows_affected(); - assert!(update == 1); - - Ok(()) + Ok(exit_status) } pub(crate) fn handle_message(&mut self, message: NixMessage) -> Result<()> { diff --git a/src/nix_util/running_verify.rs b/src/nix_util/running_verify.rs index d660fcb..4a91e27 100644 --- a/src/nix_util/running_verify.rs +++ b/src/nix_util/running_verify.rs @@ -71,9 +71,9 @@ impl RunningVerify { }; match message { NixAction::Msg(msg_message) => { - // if msg_message.level > 0 && msg_message.level < 5 { - // eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg); - // } + if msg_message.level > 0 && msg_message.level < 4 { + eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg); + } } NixAction::Start(activity_start_message) => { // println!("START: {}", serde_json::to_string(&activity_start_message)?);