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 crate::nix_util::transparent_iter::TransparentIter; use super::activity_tree::ActivityId; 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 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 { 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) => { 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 mut tree = String::new(); let draw_order = self.get_draw_order(); 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 = self.activity_tree.get_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); } self.last_announce = Some(Instant::now()); } fn get_draw_order(&self) -> Vec { let mut draw_order: Vec = Vec::new(); let mut stack: Vec = self .get_children_in_order(self.activity_tree.get_tree().get_root_id()) .filter_map(|child| { if child.get_activity().is_active() { Some(DrawStackEntry::HasNotVisitedChildren(DrawDagEntry { activity_id: child.get_activity_id().clone(), depth: Vec::new(), has_later_siblings: true, })) } else { None } }) .collect(); if stack.len() == 0 { return draw_order; } stack .last_mut() .expect("If-statement ensured this is Some()") .set_no_later_siblings(); while !stack.is_empty() { let current_entry = stack.pop().expect("While-loop ensured this is Some."); match current_entry { DrawStackEntry::HasNotVisitedChildren(draw_dag_entry) => { let current_id = draw_dag_entry.activity_id.clone(); let mut current_depth = draw_dag_entry.depth.clone(); current_depth.push(draw_dag_entry.has_later_siblings); stack.push(DrawStackEntry::VisitedChildren(draw_dag_entry)); let children: Vec = self .get_children_in_order(current_id) .filter_map(|child| { if child.get_activity().is_active() { Some(child.get_activity_id().clone()) } else { None } }) .collect(); let has_children = !children.is_empty(); stack.extend(children.into_iter().map(|child_id| { DrawStackEntry::HasNotVisitedChildren(DrawDagEntry { activity_id: child_id, depth: current_depth.clone(), has_later_siblings: true, }) })); if has_children { stack .last_mut() .expect("If-statement ensured this is Some()") .set_no_later_siblings(); } } DrawStackEntry::VisitedChildren(draw_dag_entry) => { draw_order.push(draw_dag_entry); } }; } draw_order } fn get_children_in_order( &self, parent_id: ActivityId, ) -> impl Iterator { let parent = self.activity_tree.get_tree().get(&parent_id); parent .get_child_ids() .iter() .map(|child_id| self.activity_tree.get_tree().get(child_id)) .flat_map(|child| TransparentIter::new(self.activity_tree.get_tree(), child)) } } enum DrawStackEntry { HasNotVisitedChildren(DrawDagEntry), VisitedChildren(DrawDagEntry), } struct DrawDagEntry { activity_id: ActivityId, depth: Vec, has_later_siblings: bool, } impl DrawStackEntry { fn set_no_later_siblings(&mut self) -> () { match self { DrawStackEntry::HasNotVisitedChildren(draw_dag_entry) => { draw_dag_entry.has_later_siblings = false } DrawStackEntry::VisitedChildren(draw_dag_entry) => { draw_dag_entry.has_later_siblings = false } }; } }