Support update builds.

src/command/build/runner.rs

@@ -7,6 +7,7 @@ use crate::fs_util::assert_directory;
 use crate::fs_util::is_git_repo;
 use crate::git_util::git_force_into_state;
 use crate::git_util::git_init_at_rev;
+use crate::nix_util::nix_flake_update;
 use crate::nix_util::nixos_build_target;
 
 pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
@@ -38,6 +39,9 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
         };
 
         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?;
     }
 
@@ -76,6 +80,14 @@ async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig)
     Ok(())
 }
 
+async fn run_nix_update(config_root: &Config, target_config: &TargetConfig) -> Result<()> {
+    let flake_directory = target_config.get_flake_directory(config_root)?;
+
+    nix_flake_update(flake_directory).await?;
+
+    Ok(())
+}
+
 async fn build_target(
     db_handle: &DbHandle,
     config_root: &Config,

src/command/feed_logs/runner.rs

@@ -14,10 +14,12 @@ use crate::nix_util::NixOutputStream;
 use crate::nix_util::OutputLine;
 use crate::nix_util::OutputLineStream;
 use crate::nix_util::RunningBuild;
+use crate::nix_util::RunningUpdate;
 
 pub(crate) async fn feed_logs(args: FeedLogArgs) -> Result<()> {
     let db_handle = DbHandle::new::<String>(None).await?;
-    let mut running_build = RunningBuild::new(&db_handle)?;
+    let mut running_build = RunningUpdate::new()?;
+    // let mut running_build = RunningBuild::new(&db_handle)?;
     let file_stream = FileStream::new(args.input).await?;
     let mut nix_output_stream = NixOutputStream::new(file_stream);
     while let Some(message) = nix_output_stream.next().await? {

src/config/target.rs

@@ -126,6 +126,13 @@ impl TargetConfig {
     pub(crate) fn get_name(&'_ self) -> Result<&String, CustomError> {
         Ok(&self.name)
     }
+
+    /// Whether this job should update the flake inputs before the build.
+    ///
+    /// This is used for proactively building updated packages so we can quickly update machines without waiting.
+    pub(crate) fn get_update(&self) -> bool {
+        self.update.unwrap_or(false)
+    }
 }
 
 fn hex_string(data: &[u8]) -> String {

src/nix_util/activity_tree_stream.rs

@@ -0,0 +1,274 @@
+use crate::Result;
+
+use super::activity::Activity;
+use super::activity::ActivityBuild;
+use super::activity::ActivityBuildWaiting;
+use super::activity::ActivityBuilds;
+use super::activity::ActivityCopyPath;
+use super::activity::ActivityCopyPaths;
+use super::activity::ActivityFetchTree;
+use super::activity::ActivityFileTransfer;
+use super::activity::ActivityOptimizeStore;
+use super::activity::ActivityPostBuildHook;
+use super::activity::ActivityQueryPathInfo;
+use super::activity::ActivityRealize;
+use super::activity::ActivityState;
+use super::activity::ActivitySubstitute;
+use super::activity::ActivityUnknown;
+use super::activity::ActivityVerifyPaths;
+use super::activity_tree::ActivityTree;
+use super::nix_output_stream::ActivityResultMessage;
+use super::nix_output_stream::ActivityStartMessage;
+use super::nix_output_stream::NixAction;
+use super::nix_output_stream::NixMessage;
+
+pub(crate) struct ActivityTreeStream {
+    activity_tree: ActivityTree,
+}
+
+impl ActivityTreeStream {
+    pub(crate) fn new() -> ActivityTreeStream {
+        ActivityTreeStream {
+            activity_tree: ActivityTree::new(),
+        }
+    }
+
+    pub(crate) fn get_tree(&self) -> &ActivityTree {
+        &self.activity_tree
+    }
+
+    pub(crate) fn get_tree_mut(&mut self) -> &mut ActivityTree {
+        &mut self.activity_tree
+    }
+
+    pub(crate) fn handle_message(&mut self, message: &NixMessage) -> Result<()> {
+        let message = match message {
+            NixMessage::ParseFailure(_) | NixMessage::Generic(_, _) => {
+                return Ok(());
+            }
+            NixMessage::Action(nix_action) => nix_action,
+        };
+
+        match message {
+            NixAction::Msg(_msg_message) => {
+                // if msg_message.msg.contains("nix log") {
+                //     eprintln!("{}", msg_message.msg);
+                // }
+                // if msg_message.level == 0 {
+                //     eprintln!("{}", msg_message.msg);
+                // }
+            }
+            NixAction::Start(activity_start_message) => {
+                match activity_start_message {
+                    ActivityStartMessage::Unknown(activity_start_unknown) => {
+                        self.activity_tree.add_activity(
+                            activity_start_unknown.id,
+                            activity_start_unknown.parent,
+                            Activity::Unknown(ActivityUnknown {
+                                state: ActivityState::default(),
+                                text: activity_start_unknown.text.clone(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::CopyPath(activity_start_copy_path) => {
+                        self.activity_tree.add_activity(
+                            activity_start_copy_path.id,
+                            activity_start_copy_path.parent,
+                            Activity::CopyPath(ActivityCopyPath {
+                                state: ActivityState::default(),
+                                missing_path: activity_start_copy_path.missing_path.clone(),
+                                source: activity_start_copy_path.source.clone(),
+                                destination: activity_start_copy_path.destination.clone(),
+                                done: 0,
+                                expected: 0,
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::FileTransfer(activity_start_file_transfer) => {
+                        self.activity_tree.add_activity(
+                            activity_start_file_transfer.id,
+                            activity_start_file_transfer.parent,
+                            Activity::FileTransfer(ActivityFileTransfer {
+                                state: ActivityState::default(),
+                                url: activity_start_file_transfer.url.clone(),
+                                done: 0,
+                                expected: 0,
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::Realize(activity_start_realize) => {
+                        self.activity_tree.add_activity(
+                            activity_start_realize.id,
+                            activity_start_realize.parent,
+                            Activity::Realize(ActivityRealize {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::CopyPaths(activity_start_copy_paths) => {
+                        self.activity_tree.add_activity(
+                            activity_start_copy_paths.id,
+                            activity_start_copy_paths.parent,
+                            Activity::CopyPaths(ActivityCopyPaths {
+                                state: ActivityState::default(),
+                                text: activity_start_copy_paths.text.clone(),
+                                done: 0,
+                                expected: 0,
+                                running: 0,
+                                failed: 0,
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::Builds(activity_start_builds) => {
+                        self.activity_tree.add_activity(
+                            activity_start_builds.id,
+                            activity_start_builds.parent,
+                            Activity::Builds(ActivityBuilds {
+                                state: ActivityState::default(),
+                                done: 0,
+                                expected: 0,
+                                running: 0,
+                                failed: 0,
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::Build(activity_start_build) => {
+                        self.activity_tree.add_activity(
+                            activity_start_build.id,
+                            activity_start_build.parent,
+                            Activity::Build(ActivityBuild {
+                                state: ActivityState::default(),
+                                drv_path: activity_start_build.drv_path.clone(),
+                                machine_name: if activity_start_build.machine_name.len() > 0 {
+                                    Some(activity_start_build.machine_name.clone())
+                                } else {
+                                    None
+                                },
+                                phase: None,
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::OptimizeStore(activity_start_optimize_store) => {
+                        self.activity_tree.add_activity(
+                            activity_start_optimize_store.id,
+                            activity_start_optimize_store.parent,
+                            Activity::OptimizeStore(ActivityOptimizeStore {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::VerifyPaths(activity_start_verify_paths) => {
+                        self.activity_tree.add_activity(
+                            activity_start_verify_paths.id,
+                            activity_start_verify_paths.parent,
+                            Activity::VerifyPaths(ActivityVerifyPaths {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::Substitute(activity_start_substitute) => {
+                        self.activity_tree.add_activity(
+                            activity_start_substitute.id,
+                            activity_start_substitute.parent,
+                            Activity::Substitute(ActivitySubstitute {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::QueryPathInfo(activity_start_query_path_info) => {
+                        self.activity_tree.add_activity(
+                            activity_start_query_path_info.id,
+                            activity_start_query_path_info.parent,
+                            Activity::QueryPathInfo(ActivityQueryPathInfo {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::PostBuildHook(activity_start_post_build_hook) => {
+                        self.activity_tree.add_activity(
+                            activity_start_post_build_hook.id,
+                            activity_start_post_build_hook.parent,
+                            Activity::PostBuildHook(ActivityPostBuildHook {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::BuildWaiting(activity_start_build_waiting) => {
+                        self.activity_tree.add_activity(
+                            activity_start_build_waiting.id,
+                            0,
+                            Activity::BuildWaiting(ActivityBuildWaiting {
+                                state: ActivityState::default(),
+                                text: activity_start_build_waiting.text.clone(),
+                                drv_path: activity_start_build_waiting.drv_path.clone(),
+                                path_resolved: activity_start_build_waiting.path_resolved.clone(),
+                            }),
+                        )?;
+                    }
+                    ActivityStartMessage::FetchTree(activity_start_fetch_tree) => {
+                        self.activity_tree.add_activity(
+                            activity_start_fetch_tree.id,
+                            activity_start_fetch_tree.parent,
+                            Activity::FetchTree(ActivityFetchTree {
+                                state: ActivityState::default(),
+                            }),
+                        )?;
+                    }
+                };
+            }
+            NixAction::Stop(stop_message) => {
+                let activity = self
+                    .activity_tree
+                    .get_activity_id(stop_message.id)
+                    .map(|activity_id| self.activity_tree.get_mut(&activity_id))?;
+                activity.get_mut_activity().stop();
+                // 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) => {
+                        let activity_id = self
+                            .activity_tree
+                            .get_activity_id(activity_result_set_phase.id)?;
+                        let activity = self.activity_tree.get_mut(&activity_id);
+                        activity
+                            .get_mut_activity()
+                            .set_phase(Some(activity_result_set_phase.phase.clone()));
+                    }
+                    ActivityResultMessage::Progress(activity_result_progress) => {
+                        let activity_id = self
+                            .activity_tree
+                            .get_activity_id(activity_result_progress.id)?;
+                        let activity = self.activity_tree.get_mut(&activity_id);
+                        activity.get_mut_activity().set_progress(
+                            activity_result_progress.done,
+                            activity_result_progress.expected,
+                            activity_result_progress.running,
+                            activity_result_progress.failed,
+                        );
+                    }
+                    ActivityResultMessage::SetExpected(activity_result_set_expected) => {
+                        let activity_id = self
+                            .activity_tree
+                            .get_activity_id(activity_result_set_expected.id)?;
+                        let activity = self.activity_tree.get_mut(&activity_id);
+                        activity
+                            .get_mut_activity()
+                            .set_expected(activity_result_set_expected.expected);
+                    }
+                    ActivityResultMessage::PostBuildLogLine(
+                        _activity_result_post_build_log_line,
+                    ) => {}
+                    ActivityResultMessage::FetchStatus(_activity_result_fetch_status) => {}
+                };
+            }
+        };
+        Ok(())
+    }
+}

src/nix_util/high_level.rs

@@ -8,6 +8,7 @@ use tokio::process::Command;
 use crate::Result;
 use crate::database::db_handle::DbHandle;
 
+use super::RunningUpdate;
 use super::running_build::RunningBuild;
 
 pub(crate) async fn nixos_build_target<B, F, A, TN>(
@@ -59,3 +60,29 @@ where
 
     Ok(())
 }
+
+pub(crate) async fn nix_flake_update<F>(flake_path: F) -> Result<()>
+where
+    F: AsRef<Path>,
+{
+    let mut command = Command::new("nix");
+    command.current_dir(flake_path);
+    command.stdout(Stdio::piped());
+    command.stderr(Stdio::piped());
+    command.stdin(Stdio::null());
+    command.args([
+        "flake",
+        "update",
+        "--log-format",
+        "internal-json",
+        "-vvvvvvvvvvv",
+    ]);
+    command.kill_on_drop(true);
+
+    let child = command.spawn()?;
+
+    let mut running_update = RunningUpdate::new()?;
+    running_update.run_to_completion(child).await?;
+
+    Ok(())
+}

src/nix_util/mod.rs

@@ -1,9 +1,11 @@
 mod activity;
 mod activity_tree;
+mod activity_tree_stream;
 mod high_level;
 mod nix_output_stream;
 mod output_stream;
 mod running_build;
+mod running_update;
 mod transparent_iter;
 pub(crate) use activity_tree::ActivityIdAlreadyInTreeError;
 pub(crate) use activity_tree::ActivityIdNotInTreeError;
@@ -12,3 +14,4 @@ pub(crate) use nix_output_stream::NixOutputStream;
 pub(crate) use output_stream::OutputLine;
 pub(crate) use output_stream::OutputLineStream;
 pub(crate) use running_build::RunningBuild;
+pub(crate) use running_update::RunningUpdate;

src/nix_util/running_build.rs

@@ -9,37 +9,20 @@ use tracing::error;
 
 use crate::Result;
 use crate::database::db_handle::DbHandle;
-use crate::nix_util::nix_output_stream::ActivityStartMessage;
 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::Activity;
-use super::activity::ActivityBuild;
-use super::activity::ActivityBuildWaiting;
-use super::activity::ActivityBuilds;
-use super::activity::ActivityCopyPath;
-use super::activity::ActivityCopyPaths;
-use super::activity::ActivityFetchTree;
-use super::activity::ActivityFileTransfer;
-use super::activity::ActivityOptimizeStore;
-use super::activity::ActivityPostBuildHook;
-use super::activity::ActivityQueryPathInfo;
-use super::activity::ActivityRealize;
-use super::activity::ActivityState;
-use super::activity::ActivitySubstitute;
-use super::activity::ActivityUnknown;
-use super::activity::ActivityVerifyPaths;
 use super::activity_tree::ActivityId;
-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;
 
 pub(crate) struct RunningBuild<'db> {
     db_handle: &'db DbHandle,
-    activity_tree: ActivityTree,
+    activity_tree: ActivityTreeStream,
     last_announce: Option<Instant>,
 }
 
@@ -47,7 +30,7 @@ impl<'db> RunningBuild<'db> {
     pub(crate) fn new(db_handle: &'db DbHandle) -> Result<Self> {
         Ok(RunningBuild {
             db_handle,
-            activity_tree: ActivityTree::new(),
+            activity_tree: ActivityTreeStream::new(),
             last_announce: None,
         })
     }
@@ -106,6 +89,8 @@ impl<'db> RunningBuild<'db> {
     }
 
     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}");
@@ -119,178 +104,14 @@ impl<'db> RunningBuild<'db> {
         };
         match message {
             NixAction::Msg(msg_message) => {
-                if msg_message.msg.contains("nix log") {
+                if msg_message.level > 0 && msg_message.level < 5 {
-                    eprintln!("{}", msg_message.msg);
+                    eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg);
                 }
-                // if msg_message.level == 0 {
-                //     eprintln!("{}", msg_message.msg);
-                // }
             }
-            NixAction::Start(activity_start_message) => {
+            NixAction::Start(_activity_start_message) => {
-                match activity_start_message {
-                    ActivityStartMessage::Unknown(activity_start_unknown) => {
-                        self.activity_tree.add_activity(
-                            activity_start_unknown.id,
-                            activity_start_unknown.parent,
-                            Activity::Unknown(ActivityUnknown {
-                                state: ActivityState::default(),
-                                text: activity_start_unknown.text,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::CopyPath(activity_start_copy_path) => {
-                        self.activity_tree.add_activity(
-                            activity_start_copy_path.id,
-                            activity_start_copy_path.parent,
-                            Activity::CopyPath(ActivityCopyPath {
-                                state: ActivityState::default(),
-                                missing_path: activity_start_copy_path.missing_path,
-                                source: activity_start_copy_path.source,
-                                destination: activity_start_copy_path.destination,
-                                done: 0,
-                                expected: 0,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::FileTransfer(activity_start_file_transfer) => {
-                        self.activity_tree.add_activity(
-                            activity_start_file_transfer.id,
-                            activity_start_file_transfer.parent,
-                            Activity::FileTransfer(ActivityFileTransfer {
-                                state: ActivityState::default(),
-                                url: activity_start_file_transfer.url,
-                                done: 0,
-                                expected: 0,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::Realize(activity_start_realize) => {
-                        self.activity_tree.add_activity(
-                            activity_start_realize.id,
-                            activity_start_realize.parent,
-                            Activity::Realize(ActivityRealize {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::CopyPaths(activity_start_copy_paths) => {
-                        self.activity_tree.add_activity(
-                            activity_start_copy_paths.id,
-                            activity_start_copy_paths.parent,
-                            Activity::CopyPaths(ActivityCopyPaths {
-                                state: ActivityState::default(),
-                                text: activity_start_copy_paths.text,
-                                done: 0,
-                                expected: 0,
-                                running: 0,
-                                failed: 0,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::Builds(activity_start_builds) => {
-                        self.activity_tree.add_activity(
-                            activity_start_builds.id,
-                            activity_start_builds.parent,
-                            Activity::Builds(ActivityBuilds {
-                                state: ActivityState::default(),
-                                done: 0,
-                                expected: 0,
-                                running: 0,
-                                failed: 0,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::Build(activity_start_build) => {
-                        self.activity_tree.add_activity(
-                            activity_start_build.id,
-                            activity_start_build.parent,
-                            Activity::Build(ActivityBuild {
-                                state: ActivityState::default(),
-                                drv_path: activity_start_build.drv_path,
-                                machine_name: if activity_start_build.machine_name.len() > 0 {
-                                    Some(activity_start_build.machine_name)
-                                } else {
-                                    None
-                                },
-                                phase: None,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::OptimizeStore(activity_start_optimize_store) => {
-                        self.activity_tree.add_activity(
-                            activity_start_optimize_store.id,
-                            activity_start_optimize_store.parent,
-                            Activity::OptimizeStore(ActivityOptimizeStore {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::VerifyPaths(activity_start_verify_paths) => {
-                        self.activity_tree.add_activity(
-                            activity_start_verify_paths.id,
-                            activity_start_verify_paths.parent,
-                            Activity::VerifyPaths(ActivityVerifyPaths {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::Substitute(activity_start_substitute) => {
-                        self.activity_tree.add_activity(
-                            activity_start_substitute.id,
-                            activity_start_substitute.parent,
-                            Activity::Substitute(ActivitySubstitute {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::QueryPathInfo(activity_start_query_path_info) => {
-                        self.activity_tree.add_activity(
-                            activity_start_query_path_info.id,
-                            activity_start_query_path_info.parent,
-                            Activity::QueryPathInfo(ActivityQueryPathInfo {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::PostBuildHook(activity_start_post_build_hook) => {
-                        self.activity_tree.add_activity(
-                            activity_start_post_build_hook.id,
-                            activity_start_post_build_hook.parent,
-                            Activity::PostBuildHook(ActivityPostBuildHook {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::BuildWaiting(activity_start_build_waiting) => {
-                        self.activity_tree.add_activity(
-                            activity_start_build_waiting.id,
-                            0,
-                            Activity::BuildWaiting(ActivityBuildWaiting {
-                                state: ActivityState::default(),
-                                text: activity_start_build_waiting.text,
-                                drv_path: activity_start_build_waiting.drv_path,
-                                path_resolved: activity_start_build_waiting.path_resolved,
-                            }),
-                        )?;
-                    }
-                    ActivityStartMessage::FetchTree(activity_start_fetch_tree) => {
-                        self.activity_tree.add_activity(
-                            activity_start_fetch_tree.id,
-                            activity_start_fetch_tree.parent,
-                            Activity::FetchTree(ActivityFetchTree {
-                                state: ActivityState::default(),
-                            }),
-                        )?;
-                    }
-                };
                 self.print_current_status();
             }
-            NixAction::Stop(stop_message) => {
+            NixAction::Stop(_stop_message) => {
-                let activity = self
-                    .activity_tree
-                    .get_activity_id(stop_message.id)
-                    .map(|activity_id| self.activity_tree.get_mut(&activity_id))?;
-                activity.get_mut_activity().stop();
                 self.print_current_status();
                 // println!("{}", serde_json::to_string(&message)?);
             }
@@ -302,40 +123,13 @@ impl<'db> RunningBuild<'db> {
                     }
                     ActivityResultMessage::UntrustedPath(_activity_result_untrusted_path) => {}
                     ActivityResultMessage::CorruptedPath(_activity_result_corrupted_path) => {}
-                    ActivityResultMessage::SetPhase(activity_result_set_phase) => {
+                    ActivityResultMessage::SetPhase(_activity_result_set_phase) => {
-                        let activity_id = self
-                            .activity_tree
-                            .get_activity_id(activity_result_set_phase.id)?;
-                        let activity = self.activity_tree.get_mut(&activity_id);
-                        activity
-                            .get_mut_activity()
-                            .set_phase(Some(activity_result_set_phase.phase));
-
                         self.print_current_status();
                     }
-                    ActivityResultMessage::Progress(activity_result_progress) => {
+                    ActivityResultMessage::Progress(_activity_result_progress) => {
-                        let activity_id = self
-                            .activity_tree
-                            .get_activity_id(activity_result_progress.id)?;
-                        let activity = self.activity_tree.get_mut(&activity_id);
-                        activity.get_mut_activity().set_progress(
-                            activity_result_progress.done,
-                            activity_result_progress.expected,
-                            activity_result_progress.running,
-                            activity_result_progress.failed,
-                        );
-
                         self.maybe_print_current_status();
                     }
-                    ActivityResultMessage::SetExpected(activity_result_set_expected) => {
+                    ActivityResultMessage::SetExpected(_activity_result_set_expected) => {
-                        let activity_id = self
-                            .activity_tree
-                            .get_activity_id(activity_result_set_expected.id)?;
-                        let activity = self.activity_tree.get_mut(&activity_id);
-                        activity
-                            .get_mut_activity()
-                            .set_expected(activity_result_set_expected.expected);
-
                         self.maybe_print_current_status();
                     }
                     ActivityResultMessage::PostBuildLogLine(
@@ -384,7 +178,7 @@ impl<'db> RunningBuild<'db> {
             } else {
                 "𜸛"
             };
-            let activity = self.activity_tree.get(&dag_entry.activity_id);
+            let activity = self.activity_tree.get_tree().get(&dag_entry.activity_id);
             let display_name = activity
                 .get_activity()
                 .display_name()
@@ -407,7 +201,7 @@ impl<'db> RunningBuild<'db> {
     fn get_draw_order(&self) -> Vec<DrawDagEntry> {
         let mut draw_order: Vec<DrawDagEntry> = Vec::new();
         let mut stack: Vec<DrawStackEntry> = self
-            .get_children_in_order(self.activity_tree.get_root_id())
+            .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 {
@@ -473,12 +267,12 @@ impl<'db> RunningBuild<'db> {
         &self,
         parent_id: ActivityId,
     ) -> impl Iterator<Item = &ActivityTreeEntry> {
-        let parent = self.activity_tree.get(&parent_id);
+        let parent = self.activity_tree.get_tree().get(&parent_id);
         parent
             .get_child_ids()
             .iter()
-            .map(|child_id| self.activity_tree.get(child_id))
+            .map(|child_id| self.activity_tree.get_tree().get(child_id))
-            .flat_map(|child| TransparentIter::new(&self.activity_tree, child))
+            .flat_map(|child| TransparentIter::new(self.activity_tree.get_tree(), child))
     }
 }
 

src/nix_util/running_update.rs

@@ -0,0 +1,138 @@
+use tokio::process::Child;
+use tracing::error;
+
+use crate::Result;
+use crate::nix_util::NixOutputStream;
+use crate::nix_util::nix_output_stream::NixAction;
+use crate::nix_util::output_stream::OutputStream;
+
+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;
+use super::transparent_iter::TransparentIter;
+
+pub(crate) struct RunningUpdate {
+    activity_tree: ActivityTreeStream,
+}
+
+impl RunningUpdate {
+    pub(crate) fn new() -> Result<Self> {
+        Ok(RunningUpdate {
+            activity_tree: ActivityTreeStream::new(),
+        })
+    }
+
+    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<OutputStream> =
+            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);
+
+        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) -> () {}
+
+    fn print_current_status(&mut self) -> () {
+        let nodes = self.get_children_in_order(self.activity_tree.get_tree().get_root_id());
+        println!("\n\n\nvvvvvv\n\n");
+        for n in nodes {
+            let activity = n.get_activity();
+            // if activity.is_active() {
+            println!("{:?}", activity.display_name());
+            // }
+        }
+        println!("\n\n\n^^^^^^\n\n");
+    }
+
+    fn get_children_in_order(
+        &self,
+        parent_id: ActivityId,
+    ) -> impl Iterator<Item = &ActivityTreeEntry> {
+        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))
+    }
+}