Merge of clean command.

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"

src/cli/parameters.rs

@@ -18,6 +18,9 @@ pub(crate) enum Commands {
     /// Run a single build.
     Build(BuildArgs),
 
+    /// Clean up builds.
+    Clean(CleanArgs),
+
     /// Launch a daemon to run builds.
     Daemon(DaemonArgs),
 
@@ -36,6 +39,13 @@ pub(crate) struct BuildArgs {
     pub(crate) target: Vec<String>,
 }
 
+#[derive(Args, Debug)]
+pub(crate) struct CleanArgs {
+    /// Path to the nix_builder config file.
+    #[arg(short, long)]
+    pub(crate) config: PathBuf,
+}
+
 #[derive(Args, Debug)]
 pub(crate) struct DaemonArgs {
     /// Path to the nix_builder config file.

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;
@@ -12,6 +15,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,7 +37,62 @@ 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 {
+        // 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);
+            }
+        };
+    }
+
+    db_handle.conn.close().await;
+
+    Ok(())
+}
+
+async fn full_build_target(
+    config: &Config,
+    db_handle: &DbHandle,
+    target_name: &str,
+    build_id: i64,
+) -> Result<ExitStatus> {
     let target_config = {
         let target_config = config.get_target_config(&target_name)?;
         if let Some(conf) = target_config {
@@ -47,12 +106,8 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
     if target_config.get_update() {
         run_nix_update(&config, target_config).await?;
     }
-        build_target(&db_handle, &config, target_config).await?;
+    let exit_status = build_target(&db_handle, &config, target_config, build_id).await?;
-    }
+    Ok(exit_status)
-
-    db_handle.conn.close().await;
-
-    Ok(())
 }
 
 async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig) -> Result<()> {
@@ -85,6 +140,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)?;
 
@@ -97,7 +158,8 @@ async fn build_target(
     db_handle: &DbHandle,
     config_root: &Config,
     target_config: &TargetConfig,
-) -> Result<()> {
+    build_id: i64,
+) -> Result<ExitStatus> {
     let flake_directory = target_config.get_flake_directory(config_root)?;
     let build_directory = target_config.get_build_directory(config_root)?;
     assert_directory!(
@@ -108,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,
@@ -129,7 +183,7 @@ async fn build_target(
     )
     .await?;
 
-    Ok(())
+    Ok(exit_status)
 }
 
 async fn write_input_revs_to_db(

src/command/clean/mod.rs

@@ -0,0 +1,3 @@
+mod runner;
+
+pub(crate) use runner::run_clean;

src/command/clean/runner.rs

@@ -0,0 +1,80 @@
+use std::collections::HashSet;
+
+use tracing::info;
+
+use crate::Result;
+use crate::cli::parameters::CleanArgs;
+use crate::config::Config;
+
+pub(crate) async fn run_clean(args: CleanArgs) -> Result<()> {
+    let config = Config::load_from_file(args.config).await?;
+    clean_targets(&config).await?;
+    clean_repos(&config).await?;
+    Ok(())
+}
+
+async fn clean_targets(config: &Config) -> Result<()> {
+    let target_directory = config.get_target_root_directory()?;
+    let existing_target_names = {
+        let mut existing_entries = tokio::fs::read_dir(&target_directory).await?;
+        let mut existing_target_names = HashSet::new();
+        while let Some(existing_target) = existing_entries.next_entry().await? {
+            existing_target_names.insert(existing_target.file_name());
+        }
+        existing_target_names
+    };
+
+    let configured_target_names = config
+        .get_target_configs()
+        .iter()
+        .map(|target| {
+            target
+                .get_target_directory(&config)
+                .and_then(|target_path| {
+                    target_path
+                        .file_name()
+                        .map(|name| name.to_os_string())
+                        .ok_or("Entry with no file name in target directory.".into())
+                })
+        })
+        .collect::<Result<HashSet<_>>>()?;
+
+    for dir in existing_target_names.difference(&configured_target_names) {
+        info!("DELETE TARGET: {:?}", dir);
+        tokio::fs::remove_dir_all(target_directory.join(dir)).await?;
+    }
+
+    Ok(())
+}
+
+async fn clean_repos(config: &Config) -> Result<()> {
+    let repo_directory = config.get_repo_root_directory()?;
+    let existing_repo_names = {
+        let mut existing_entries = tokio::fs::read_dir(&repo_directory).await?;
+        let mut existing_repo_names = HashSet::new();
+        while let Some(existing_repo) = existing_entries.next_entry().await? {
+            existing_repo_names.insert(existing_repo.file_name());
+        }
+        existing_repo_names
+    };
+
+    let configured_repo_names = config
+        .get_target_configs()
+        .iter()
+        .map(|target| {
+            target.get_repo_directory(&config).and_then(|repo_path| {
+                repo_path
+                    .file_name()
+                    .map(|name| name.to_os_string())
+                    .ok_or("Entry with no file name in repo directory.".into())
+            })
+        })
+        .collect::<Result<HashSet<_>>>()?;
+
+    for dir in existing_repo_names.difference(&configured_repo_names) {
+        info!("DELETE REPO: {:?}", dir);
+        tokio::fs::remove_dir_all(repo_directory.join(dir)).await?;
+    }
+
+    Ok(())
+}

src/command/mod.rs

@@ -1,3 +1,4 @@
 pub(crate) mod build;
+pub(crate) mod clean;
 pub(crate) mod daemon;
 pub(crate) mod feed_logs;

src/config/root.rs

@@ -63,6 +63,10 @@ impl Config {
         Ok(target_config)
     }
 
+    pub(crate) fn get_target_configs(&self) -> &Vec<TargetConfig> {
+        &self.targets
+    }
+
     /// The root directory where all the output is stored.
     pub(crate) fn get_output_directory(&self) -> Result<Cow<'_, Path>, CustomError> {
         let maybe_repo_directory = self.output_directory.as_deref().map(Cow::Borrowed);
@@ -80,4 +84,20 @@ impl Config {
         let database_path = output_directory.join("nix_builder.sqlite");
         Ok(Cow::Owned(database_path))
     }
+
+    /// The folder that contains the folders for the build targets.
+    pub(crate) fn get_target_root_directory<'target>(
+        &'target self,
+    ) -> Result<Cow<'target, Path>, CustomError> {
+        let output_directory = self.get_output_directory()?;
+        Ok(Cow::Owned(output_directory.join("target")))
+    }
+
+    /// The folder that contains the folders containing the checkouts of the git repositories.
+    pub(crate) fn get_repo_root_directory<'target>(
+        &'target self,
+    ) -> Result<Cow<'target, Path>, CustomError> {
+        let output_directory = self.get_output_directory()?;
+        Ok(Cow::Owned(output_directory.join("repo")))
+    }
 }

src/config/target.rs

@@ -36,33 +36,15 @@ pub(crate) struct TargetConfig {
 }
 
 impl TargetConfig {
-    /// The folder that contains the folders for the build targets.
-    pub(crate) fn get_target_root_directory<'target, 'root>(
-        &'target self,
-        config_root: &'root Config,
-    ) -> Result<Cow<'target, Path>, CustomError> {
-        let output_directory = config_root.get_output_directory()?;
-        Ok(Cow::Owned(output_directory.join("target")))
-    }
-
     /// The root directory for this specific build target.
     pub(crate) fn get_target_directory<'target, 'root>(
         &'target self,
         config_root: &'root Config,
     ) -> Result<Cow<'target, Path>, CustomError> {
-        let target_root = self.get_target_root_directory(config_root)?;
+        let target_root = config_root.get_target_root_directory()?;
         Ok(Cow::Owned(target_root.join(&self.name)))
     }
 
-    /// The folder that contains the folders containing the checkouts of the git repositories.
-    pub(crate) fn get_repo_root_directory<'target, 'root>(
-        &'target self,
-        config_root: &'root Config,
-    ) -> Result<Cow<'target, Path>, CustomError> {
-        let output_directory = config_root.get_output_directory()?;
-        Ok(Cow::Owned(output_directory.join("repo")))
-    }
-
     /// The root of the checkout of the git repository.
     pub(crate) fn get_repo_directory<'target, 'root>(
         &'target self,
@@ -71,7 +53,7 @@ impl TargetConfig {
         let hashed_repo = Sha256::digest(&self.repo);
         let hashed_repo = hex_string(&hashed_repo);
 
-        let repo_root = self.get_repo_root_directory(config_root)?;
+        let repo_root = config_root.get_repo_root_directory()?;
         Ok(Cow::Owned(repo_root.join(hashed_repo)))
     }
 

src/main.rs

@@ -9,6 +9,7 @@ use clap::Parser;
 use self::cli::parameters::Cli;
 use self::cli::parameters::Commands;
 use self::command::build::run_build;
+use self::command::clean::run_clean;
 use self::command::daemon::start_daemon;
 use self::command::feed_logs::feed_logs;
 use self::error::CustomError;
@@ -38,6 +39,9 @@ async fn main_body() -> Result<ExitCode> {
         Commands::Build(args) => {
             run_build(args).await?;
         }
+        Commands::Clean(args) => {
+            run_clean(args).await?;
+        }
         Commands::Daemon(args) => {
             start_daemon(args).await?;
         }

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;
@@ -9,6 +10,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<B, F, A, TN>(
@@ -18,7 +20,7 @@ pub(crate) async fn nixos_build_target<B, F, A, TN>(
     attr: A,
     target_name: TN,
     build_id: i64,
-) -> Result<()>
+) -> Result<ExitStatus>
 where
     B: AsRef<Path>,
     F: AsRef<Path>,
@@ -57,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<F>(flake_path: F) -> Result<()>
@@ -87,3 +89,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(())
+}

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;

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<ExitStatus>
     where
         TN: AsRef<str>,
     {
@@ -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 =
+        Ok(exit_status)
-            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(())
     }
 
     pub(crate) fn handle_message(&mut self, message: NixMessage) -> Result<()> {

src/nix_util/running_update.rs

@@ -1,5 +1,4 @@
 use std::borrow::Cow;
-use std::ops::Deref;
 use std::time::Duration;
 use std::time::Instant;
 
@@ -12,7 +11,6 @@ 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::Activity;
 use super::activity_tree::ActivityTreeEntry;
 use super::activity_tree_stream::ActivityTreeStream;
 use super::nix_output_stream::ActivityResultMessage;
@@ -49,7 +47,7 @@ impl RunningUpdate {
         }
 
         let exit_status = exit_status_handle.await?;
-        println!("nix build status was: {}", exit_status);
+        println!("nix update status was: {}", exit_status);
 
         Ok(())
     }
@@ -171,25 +169,8 @@ fn is_match_predicate(entry: &ActivityTreeEntry) -> bool {
     entry.get_activity().get_progress_text().is_some()
 }
 
-pub(crate) fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool {
+fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool {
     return false;
-    // 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 {

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<Instant>,
+}
+
+impl RunningVerify {
+    pub(crate) fn new() -> Result<Self> {
+        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<OutputStream> =
+            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 < 4 {
+                    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()
+}