feat(hydro_deploy): improve progress UX by collapsing nested groups (#…

…1411)

Now, when a group only has a single active task, we skip printing a line
for the group itself and instead collapse its information into the line
for the inner task (recursively as necessary). This allows us to show
more fine grained progress without overflowing the console.

hydro_deploy/core/src/deployment.rs

@@ -60,7 +60,7 @@ impl Deployment {
     pub async fn deploy(&mut self) -> Result<()> {
         self.services.retain(|weak| weak.strong_count() > 0);
 
-        progress::ProgressTracker::with_group("deploy", None, || async {
+        progress::ProgressTracker::with_group("deploy", Some(3), || async {
             let mut resource_batch = super::ResourceBatch::new();
 
             for service in self.services.iter().filter_map(Weak::upgrade) {
@@ -72,7 +72,7 @@ impl Deployment {
             }
 
             let resource_result = Arc::new(
-                progress::ProgressTracker::with_group("provision", None, || async {
+                progress::ProgressTracker::with_group("provision", Some(1), || async {
                     resource_batch
                         .provision(&mut self.resource_pool, self.last_resource_result.clone())
                         .await
@@ -85,12 +85,16 @@ impl Deployment {
                 host.provision(&resource_result);
             }
 
-            progress::ProgressTracker::with_group("deploy", None, || {
-                let services_future = self
-                    .services
+            let upgraded_services = self
+                .services
+                .iter()
+                .filter_map(Weak::upgrade)
+                .collect::<Vec<_>>();
+
+            progress::ProgressTracker::with_group("prepare", Some(upgraded_services.len()), || {
+                let services_future = upgraded_services
                     .iter()
-                    .filter_map(Weak::upgrade)
-                    .map(|service: Arc<RwLock<dyn Service>>| {
+                    .map(|service: &Arc<RwLock<dyn Service>>| {
                         let resource_result = &resource_result;
                         async move { service.write().await.deploy(resource_result).await }
                     })
@@ -102,13 +106,14 @@ impl Deployment {
             })
             .await?;
 
-            progress::ProgressTracker::with_group("ready", None, || {
-                let all_services_ready = self.services.iter().filter_map(Weak::upgrade).map(
-                    |service: Arc<RwLock<dyn Service>>| async move {
-                        service.write().await.ready().await?;
-                        Ok(()) as Result<()>
-                    },
-                );
+            progress::ProgressTracker::with_group("ready", Some(upgraded_services.len()), || {
+                let all_services_ready =
+                    upgraded_services
+                        .iter()
+                        .map(|service: &Arc<RwLock<dyn Service>>| async move {
+                            service.write().await.ready().await?;
+                            Ok(()) as Result<()>
+                        });
 
                 futures::future::try_join_all(all_services_ready)
             })

hydro_deploy/core/src/hydroflow_crate/build.rs

@@ -87,7 +87,7 @@ pub async fn build_crate_memoized(params: BuildParams) -> Result<&'static BuildO
         .get_or_init(MemoMap::new)
         .get_or_insert(&params, Default::default)
         .get_or_try_init(move || {
-            ProgressTracker::rich_leaf("build".to_string(), move |_, set_msg| async move {
+            ProgressTracker::rich_leaf("build", move |set_msg| async move {
                 tokio::task::spawn_blocking(move || {
                     let mut command = Command::new("cargo");
                     command.args(["build"]);

hydro_deploy/core/src/hydroflow_crate/service.rs

@@ -197,13 +197,12 @@ impl Service for HydroflowCrateService {
         }
 
         ProgressTracker::with_group(
-            &self
-                .display_id
+            self.display_id
                 .clone()
                 .unwrap_or_else(|| format!("service/{}", self.id)),
             None,
             || async {
-                let built = self.build().await?;
+                let built = ProgressTracker::leaf("build", self.build()).await?;
 
                 let host = &self.on;
                 let launched = host.provision(resource_result);
@@ -223,8 +222,7 @@ impl Service for HydroflowCrateService {
         }
 
         ProgressTracker::with_group(
-            &self
-                .display_id
+            self.display_id
                 .clone()
                 .unwrap_or_else(|| format!("service/{}", self.id)),
             None,
@@ -259,7 +257,7 @@ impl Service for HydroflowCrateService {
                 binary.stdin().send(format!("{formatted_bind_config}\n"))?;
 
                 let ready_line = ProgressTracker::leaf(
-                    "waiting for ready".to_string(),
+                    "waiting for ready",
                     tokio::time::timeout(Duration::from_secs(60), stdout_receiver),
                 )
                 .await??;
@@ -317,8 +315,7 @@ impl Service for HydroflowCrateService {
 
     async fn stop(&mut self) -> Result<()> {
         ProgressTracker::with_group(
-            &self
-                .display_id
+            self.display_id
                 .clone()
                 .unwrap_or_else(|| format!("service/{}", self.id)),
             None,
@@ -327,7 +324,7 @@ impl Service for HydroflowCrateService {
                 launched_binary.stdin().send("stop\n".to_string())?;
 
                 let timeout_result = ProgressTracker::leaf(
-                    "waiting for exit".to_owned(),
+                    "waiting for exit",
                     tokio::time::timeout(Duration::from_secs(60), launched_binary.wait()),
                 )
                 .await;