chore: use shell-candy to incrementally handle logs (#1415)

fixes #1414 by using `shell-candy` for `rover dev` and `cargo xtask`
commands, abstracting away the logic for processing individual log lines
and other output from shell commands.

the biggest impact this PR has is on our `xtask` commands - we used to
read all the log lines into memory and dump it all out at once, whereas
now we print the lines out as soon as we receive them, and can also keep
them stored for analysis after the command has completed.

Cargo.toml

@@ -119,6 +119,7 @@ serde = "1.0"
 serde_json = "1.0"
 serde_json_traversal = "0.2"
 serde_yaml = "0.9"
+shell-candy = "0.4"
 strsim = "0.10"
 strum = "0.24"
 strum_macros = "0.24"
@@ -182,6 +183,7 @@ semver = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
+shell-candy = { workspace = true }
 sputnik = { workspace = true }
 strsim = { workspace = true }
 strum = { workspace = true }

src/command/dev/mod.rs

@@ -13,9 +13,6 @@ mod schema;
 #[cfg(feature = "composition-js")]
 mod protocol;
 
-#[cfg(feature = "composition-js")]
-mod command;
-
 #[cfg(feature = "composition-js")]
 mod netstat;
 

src/command/dev/router.rs

@@ -1,15 +1,14 @@
 use anyhow::{anyhow, Context};
 use apollo_federation_types::config::RouterVersion;
 use camino::Utf8PathBuf;
-use crossbeam_channel::bounded as sync_channel;
 use reqwest::blocking::Client;
 use rover_std::{Emoji, Fs, Style};
 use semver::Version;
+use shell_candy::{ShellTask, ShellTaskBehavior, ShellTaskLog};
 
 use std::net::SocketAddr;
 use std::time::{Duration, Instant};
 
-use crate::command::dev::command::{BackgroundTask, BackgroundTaskLog};
 use crate::command::dev::do_dev::log_err_and_continue;
 use crate::command::dev::DEV_ROUTER_VERSION;
 use crate::command::install::Plugin;
@@ -26,7 +25,7 @@ pub struct RouterRunner {
     router_socket_addr: SocketAddr,
     override_install_path: Option<Utf8PathBuf>,
     client_config: StudioClientConfig,
-    router_handle: Option<BackgroundTask>,
+    router_handle: Option<ShellTask>,
     plugin_exe: Option<Utf8PathBuf>,
 }
 
@@ -102,12 +101,10 @@ impl RouterRunner {
         let mut ready = false;
         let now = Instant::now();
         let seconds = 5;
+        let endpoint = format!("http://{}/?query={{__typename}}", &self.router_socket_addr);
         while !ready && now.elapsed() < Duration::from_secs(seconds) {
             let _ = client
-                .get(format!(
-                    "http://{}/?query={{__typename}}",
-                    &self.router_socket_addr
-                ))
+                .get(&endpoint)
                 .header("Content-Type", "application/json")
                 .send()
                 .and_then(|r| r.error_for_status())
@@ -169,35 +166,34 @@ impl RouterRunner {
             let client = self.client_config.get_reqwest_client()?;
             self.write_router_config()?;
             self.maybe_install_router()?;
-            let (router_log_sender, router_log_receiver) = sync_channel(0);
-            self.router_handle = Some(BackgroundTask::new(
-                self.get_command_to_spawn()?,
-                router_log_sender,
-            )?);
-            rayon::spawn(move || loop {
-                if let Ok(BackgroundTaskLog::Stdout(stdout)) = router_log_receiver.recv() {
-                    if let Ok(stdout) = serde_json::from_str::<serde_json::Value>(&stdout) {
-                        let fields = &stdout["fields"];
-                        if let Some(level) = stdout["level"].as_str() {
-                            if let Some(message) = fields["message"].as_str() {
-                                let warn_prefix = Style::WarningPrefix.paint("WARN:");
-                                let error_prefix = Style::ErrorPrefix.paint("ERROR:");
-                                if let Some(router_span) = stdout["target"].as_str() {
-                                    match level {
-                                        "INFO" => tracing::info!(%message, %router_span),
-                                        "DEBUG" => tracing::debug!(%message, %router_span),
-                                        "TRACE" => tracing::trace!(%message, %router_span),
-                                        "WARN" => eprintln!("{} {}", warn_prefix, &message),
-                                        "ERROR" => {
-                                            eprintln!("{} {}", error_prefix, &message)
+            let router_handle = ShellTask::new(&self.get_command_to_spawn()?)?;
+            rayon::spawn(move || {
+                let _ = router_handle.run(|line| {
+                    if let ShellTaskLog::Stdout(stdout) = line {
+                        if let Ok(stdout) = serde_json::from_str::<serde_json::Value>(&stdout) {
+                            let fields = &stdout["fields"];
+                            if let Some(level) = stdout["level"].as_str() {
+                                if let Some(message) = fields["message"].as_str() {
+                                    let warn_prefix = Style::WarningPrefix.paint("WARN:");
+                                    let error_prefix = Style::ErrorPrefix.paint("ERROR:");
+                                    if let Some(router_span) = stdout["target"].as_str() {
+                                        match level {
+                                            "INFO" => tracing::info!(%message, %router_span),
+                                            "DEBUG" => tracing::debug!(%message, %router_span),
+                                            "TRACE" => tracing::trace!(%message, %router_span),
+                                            "WARN" => eprintln!("{} {}", warn_prefix, &message),
+                                            "ERROR" => {
+                                                eprintln!("{} {}", error_prefix, &message)
+                                            }
+                                            _ => {}
                                         }
-                                        _ => {}
                                     }
                                 }
                             }
                         }
                     }
-                }
+                    ShellTaskBehavior::<()>::Passthrough
+                });
             });
             self.wait_for_startup(client)
         } else {

xtask/Cargo.toml

@@ -24,6 +24,7 @@ regex = { workspace = true }
 reqwest = { workspace = true, default-features = false, features = ["blocking", "native-tls"] }
 semver = { workspace = true }
 serde_json_traversal = { workspace = true }
+shell-candy = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true, default-features = false, features = ["rt"] }
 tokio-stream = { workspace = true }

xtask/src/commands/unit_test.rs

@@ -23,11 +23,10 @@ impl UnitTest {
         if let Target::LinuxUnknownGnu = self.target {
             if env::var_os("CHECK_GLIBC").is_some() {
                 let check_glibc_script = "./check_glibc.sh".to_string();
-                let runner = Runner {
+                let runner = Runner::new(
+                    Utf8PathBuf::from_str(&check_glibc_script)?.as_str(),
                     verbose,
-                    tool_name: check_glibc_script.clone(),
-                    tool_exe: Utf8PathBuf::from_str(&check_glibc_script)?,
-                };
+                );
                 let bin_path = format!("./target/{}/debug/rover", &self.target);
                 runner.exec(&[&bin_path], &PKG_PROJECT_ROOT, None)?;
             }

xtask/src/tools/cargo.rs

@@ -16,7 +16,7 @@ pub(crate) struct CargoRunner {
 impl CargoRunner {
     /// Creates a new cargo runner with knowledge of the root rover binary and all plugins
     pub(crate) fn new(verbose: bool) -> Result<Self> {
-        let runner = Runner::new("cargo", verbose)?;
+        let runner = Runner::new("cargo", verbose);
         Ok(CargoRunner {
             cargo_package_directory: PKG_PROJECT_ROOT.clone(),
             runner,

xtask/src/tools/git.rs

@@ -29,14 +29,14 @@ struct TmpRepo {
 
 impl GitRunner {
     pub(crate) fn new(verbose: bool, path: &Utf8PathBuf) -> Result<Self> {
-        let runner = Runner::new("git", verbose)?;
+        let runner = Runner::new("git", verbose);
         Ok(GitRunner {
             runner,
             repo: RepoLocation::Local(LocalRepo { path: path.clone() }),
         })
     }
     pub(crate) fn tmp(verbose: bool) -> Result<Self> {
-        let runner = Runner::new("git", verbose)?;
+        let runner = Runner::new("git", verbose);
         let temp_dir = TempDir::new().with_context(|| "Could not create temp directory")?;
         let temp_dir_path = Utf8PathBuf::try_from(temp_dir.path().to_path_buf())
             .with_context(|| "Temp directory was not valid Utf-8")?;

xtask/src/tools/make.rs

@@ -13,7 +13,7 @@ pub(crate) struct MakeRunner {
 
 impl MakeRunner {
     pub(crate) fn new(verbose: bool, rover_exe: Utf8PathBuf) -> Result<Self> {
-        let runner = Runner::new("make", verbose)?;
+        let runner = Runner::new("make", verbose);
 
         Ok(MakeRunner { runner, rover_exe })
     }
@@ -48,7 +48,7 @@ fn assert_demo_includes(output: &CommandOutput) -> Result<()> {
         Ok(())
     } else {
         Err(anyhow!(
-            "The output from 'make` is missing the following strings: {:?}",
+            "The output from `make` is missing the following strings: {:?}",
             missing_strings
         ))
     }

xtask/src/tools/npm.rs

@@ -20,7 +20,7 @@ pub(crate) struct NpmRunner {
 
 impl NpmRunner {
     pub(crate) fn new(verbose: bool) -> Result<Self> {
-        let runner = Runner::new("npm", verbose)?;
+        let runner = Runner::new("npm", verbose);
         let project_root = PKG_PROJECT_ROOT.clone();
 
         let rover_client_lint_directory = project_root.join("crates").join("rover-client");
@@ -136,7 +136,7 @@ impl NpmRunner {
             Ok(())
         } else {
             Err(anyhow!(
-                "$FLYBY_APOLLO_KEY is not set and this does not appear to be a forked PR..."
+                "$FLYBY_APOLLO_KEY is not set and this does not appear to be a forked PR. This API key should have permissions to run checks on the `flyby-rover` graph (https://studio.apollographql.com/graph/flyby-rover) and it can be set in ./examples/flyby/.env."
             ))
         }
     }