use std::borrow::Cow; use std::time::Duration; use std::time::Instant; use std::time::SystemTime; use sqlx::Row; use tokio::process::Child; use tracing::error; use crate::Result; use crate::database::db_handle::DbHandle; use crate::nix_util::nix_output_stream::NixAction; use crate::nix_util::nix_output_stream::NixOutputStream; use crate::nix_util::output_stream::OutputStream; use super::activity::Activity; use super::activity_tree::ActivityTree; use super::activity_tree::ActivityTreeEntry; use super::activity_tree_stream::ActivityTreeStream; use super::nix_output_stream::ActivityResultMessage; use super::nix_output_stream::NixMessage; use super::tree_iter::DrawDagEntry; use super::tree_iter::ReverseTreeIter; pub(crate) struct RunningBuild<'db> { db_handle: &'db DbHandle, activity_tree: ActivityTreeStream, last_announce: Option, } impl<'db> RunningBuild<'db> { pub(crate) fn new(db_handle: &'db DbHandle) -> Result { Ok(RunningBuild { db_handle, activity_tree: ActivityTreeStream::new(), last_announce: None, }) } pub(crate) async fn run_to_completion( &mut self, mut child: Child, target_name: TN, ) -> Result<()> where TN: AsRef, { let foo = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?; let now = Instant::now(); let build_id: i64 = sqlx::query( r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#, ) .bind(target_name.as_ref()) .fetch_one(&self.db_handle.conn) .await? .try_get("id")?; let output_stream = OutputStream::from_child(&mut child)?; let mut nix_output_stream: NixOutputStream = NixOutputStream::new(output_stream); 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?; 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(build_id) .execute(&self.db_handle.conn) .await? .rows_affected(); assert!(update == 1); 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 && !(msg_message.level == 4 && msg_message.msg.starts_with("linking ")) { eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg); } } NixAction::Start(_activity_start_message) => { self.print_current_status(); } NixAction::Stop(_stop_message) => { self.print_current_status(); // println!("{}", serde_json::to_string(&message)?); } NixAction::Result(activity_result_message) => { match activity_result_message { ActivityResultMessage::FileLinked(_activity_result_file_linked) => {} ActivityResultMessage::BuildLogLine(_activity_result_build_log_line) => { // These are the output from the actual build (as opposed to the output from nix). } ActivityResultMessage::UntrustedPath(_activity_result_untrusted_path) => {} ActivityResultMessage::CorruptedPath(_activity_result_corrupted_path) => {} ActivityResultMessage::SetPhase(_activity_result_set_phase) => { self.print_current_status(); } ActivityResultMessage::Progress(activity_result_progress) => { if activity_result_progress.done == activity_result_progress.expected && activity_result_progress.expected > 0 { // If we finished the in-progress activity, always render so we show the 100% done version. self.print_current_status(); } else { self.maybe_print_current_status(); } } ActivityResultMessage::SetExpected(_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 draw_order = get_draw_order(self.activity_tree.get_tree()); let draw_order = ReverseTreeIter::new( self.activity_tree.get_tree(), None, is_match_predicate, is_transparent_predicate, is_alive_predicate, ) .get_draw_order(); print_dag(self.activity_tree.get_tree(), draw_order); self.last_announce = Some(Instant::now()); } } pub(crate) fn print_dag(activity_tree: &ActivityTree, draw_order: Vec) -> () { let mut tree = String::new(); for dag_entry in draw_order { let leading_bars = { let mut leading_bars = String::with_capacity(3 * dag_entry.depth.len()); for leading_bar in dag_entry .depth .iter() .map(|depth| if *depth { "𜹈 " } else { " " }) { leading_bars.push_str(leading_bar); } leading_bars }; let branch = if dag_entry.has_later_siblings { "𜸨" } else { "𜸛" }; let activity = activity_tree.get(&dag_entry.activity_id); let display_name = activity .get_activity() .display_name() .expect("Currently we always return a display name."); let progress_text = activity.get_activity().get_progress_text(); let (progress, progress_sep) = match progress_text { Some(text) => (text, " "), None => (Cow::Borrowed(""), ""), }; tree += &format!("{leading_bars}{branch}𜸟 {progress}{progress_sep}{display_name}\n"); } if tree.is_empty() { println!("No active activities."); } else { print!("\n{}\n", tree); } } fn is_match_predicate(_entry: &ActivityTreeEntry) -> bool { true } pub(crate) fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool { match entry.get_activity() { Activity::Root(_activity_root) => true, Activity::Unknown(_activity_unknown) => false, Activity::CopyPath(_activity_copy_path) => false, Activity::FileTransfer(_activity_file_transfer) => false, Activity::Realize(_activity_realize) => true, Activity::CopyPaths(_activity_copy_paths) => true, Activity::Builds(_activity_builds) => true, Activity::Build(_activity_build) => false, Activity::OptimizeStore(_activity_optimize_store) => false, Activity::VerifyPaths(_activity_verify_paths) => false, Activity::Substitute(_activity_substitute) => false, Activity::QueryPathInfo(_activity_query_path_info) => false, Activity::PostBuildHook(_activity_post_build_hook) => false, Activity::BuildWaiting(_activity_build_waiting) => true, Activity::FetchTree(_activity_fetch_tree) => false, } } fn is_alive_predicate(entry: &ActivityTreeEntry) -> bool { entry.get_activity().is_active() }