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() +}