xtask: migrating from duct to xshell

xshell does not handle async, so it also requires the addition of Tokio.
process group IDs are used to kill all xtask subprocesses

Author: gabriele-0201

Cargo.lock

@@ -41,7 +41,7 @@ futures = { version = "0.3.29" }
 jsonrpsee = { version = "0.20.3" }
 tracing = { version = "0.1.40" }
 tracing-subscriber = { version = "0.3.18" }
-tokio = { version = "1.34.0" }
+tokio = { version = "1.36.0" }
 async-trait = { version = "0.1.74" }
 fex = { version = "0.4.3" }
 hex-literal = { version = "0.4.1" }
@@ -184,5 +184,6 @@ pallet-ikura-blobs = { path = "ikura/chain/pallets/blobs", default-features = fa
 pallet-ikura-length-fee-adjustment = { path = "ikura/chain/pallets/length-fee-adjustment", default-features = false }
 
 # xtask
-duct = { version = "0.13.7" }
+xshell = { version = "0.2.5" }
+nix = { version = "0.27.1" }
 ctrlc = { version = "3.4.2" }

Cargo.toml

@@ -10,9 +10,11 @@ license.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-duct = { workspace = true }
+xshell = { workspace = true }
+nix = { workspace = true, features = ["signal", "process"] }
+tokio = { workspace = true, features = ["rt", "macros", "rt-multi-thread", "time", "process", "sync", "signal"] }
 clap = { workspace = true, features = ["derive"] }
 anyhow = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
-ctrlc = { workspace = true }
+serde_json = { workspace = true }

xtask/Cargo.toml

@@ -1,44 +1,37 @@
-use crate::{cli::BuildParams, logging::create_with_logs};
-use duct::cmd;
+use crate::{
+    cli::BuildParams,
+    logging::{create_log_file, WithLogs},
+};
+use xshell::cmd;
 
 // TODO: https://github.com/thrumdev/blobs/issues/225
 
-pub fn build(project_path: &std::path::Path, params: BuildParams) -> anyhow::Result<()> {
+pub async fn build(project_path: &std::path::Path, params: BuildParams) -> anyhow::Result<()> {
     if params.skip {
         return Ok(());
     }
 
+    let sh = xshell::Shell::new()?;
+
     tracing::info!("Building logs redirected {}", params.log_path);
-    let with_logs = create_with_logs(project_path, params.log_path);
+    let log_path = create_log_file(project_path, &params.log_path);
 
     // `it is advisable to use CARGO environmental variable to get the right cargo`
     // quoted by xtask readme
     let cargo = std::env::var("CARGO").unwrap_or_else(|_| "cargo".to_string());
 
-    with_logs(
-        "Building ikura-node",
-        cmd!(&cargo, "build", "-p", "ikura-node", "--release"),
-    )
-    .run()?;
-
-    with_logs(
-        "Building ikura-shim",
-        cmd!(&cargo, "build", "-p", "ikura-shim", "--release"),
-    )
-    .run()?;
-
-    let sov_demo_rollup_path = project_path.join("demo/sovereign/demo-rollup/");
-    #[rustfmt::skip]
-    with_logs(
-        "Building sovereign demo-rollup",
-        cmd!(
-            "sh", "-c",
-            format!(
-                "cd {}  && {cargo} build --release",
-                sov_demo_rollup_path.to_string_lossy()
-            )
-        ),
-    ).run()?;
+    cmd!(sh, "{cargo} build -p ikura-node --release")
+        .run_with_logs("Building ikura-node", &log_path)
+        .await??;
+
+    cmd!(sh, "{cargo} build -p ikura-node --release")
+        .run_with_logs("Building ikura-shim", &log_path)
+        .await??;
+
+    sh.change_dir(project_path.join("demo/sovereign/demo-rollup/"));
+    cmd!(sh, "{cargo} build --release")
+        .run_with_logs("Building sovereign demo-rollup", &log_path)
+        .await??;
 
     Ok(())
 }

xtask/src/build.rs

@@ -1,65 +1,166 @@
-use std::path::Path;
-use std::{io::Write, path::PathBuf};
-use tracing::{info, warn};
+use std::{
+    fs::{create_dir_all, File},
+    io::Write,
+    os::unix::process::CommandExt,
+    path::{Path, PathBuf},
+    process::{Command as StdCommand, Stdio},
+};
+use tokio::{
+    process::{Child, Command as TokioCommand},
+    task::JoinHandle,
+};
+use tracing::warn;
 
 // If log_path is relative it will be made absolute relative to the project_path
 //
 // The absolute path of where the log file is created is returned
-fn create_log_file(project_path: &Path, log_path: &String) -> std::io::Result<PathBuf> {
+pub fn create_log_file(project_path: &Path, log_path: &String) -> Option<PathBuf> {
     let mut log_path: PathBuf = Path::new(&log_path).to_path_buf();
 
     if log_path.is_relative() {
         log_path = project_path.join(log_path);
     }
 
     if let Some(prefix) = log_path.parent() {
-        std::fs::create_dir_all(prefix)?;
+        create_dir_all(prefix)
+            .map_err(|e| warn!("Impossible to redirect logs, using stdout instead. Error: {e}"))
+            .ok()?;
     }
-    std::fs::File::create(&log_path)?;
-    Ok(log_path)
+
+    File::create(&log_path)
+        .map_err(|e| warn!("Impossible to redirect logs, using stdout instead. Error: {e}"))
+        .ok()?;
+
+    Some(log_path)
 }
 
-// If the log file cannot be created due to any reasons,
-// such as lack of permission to create files or new folders in the path,
-// things will be printed to stdout instead of being redirected to the logs file
+// These methods will accept a command description and a log path.
 //
-// The returned closure will accept a description of the command and the command itself as a duct::Expression.
-// The description will be printed to both stdout and the log file, if possible, while
-// to the expression will be added the redirection of the logs, if possible.
-pub fn create_with_logs(
-    project_path: &Path,
-    log_path: String,
-) -> Box<dyn Fn(&str, duct::Expression) -> duct::Expression> {
-    let without_logs = |description: &str, cmd: duct::Expression| -> duct::Expression {
-        info!("{description}");
-        cmd
-    };
-
-    let log_path = match create_log_file(project_path, &log_path) {
-        Ok(log_path) => log_path,
-        Err(e) => {
-            warn!("Impossible redirect logs, using stdout instead. Error: {e}");
-            return Box::new(without_logs);
-        }
-    };
-
-    let with_logs = move |description: &str, cmd: duct::Expression| -> duct::Expression {
-        // The file has just been created
-        let mut log_file = std::fs::File::options()
-            .append(true)
-            .open(&log_path)
-            .unwrap();
-
-        info!("{description}");
-        let log_path = log_path.to_string_lossy();
-        let _ = log_file
-            .write(format!("{}\n", description).as_bytes())
-            .map_err(|e| warn!("Error writing into {log_path}, error: {e}",));
-        let _ = log_file
-            .flush()
-            .map_err(|e| warn!("Error writing into {log_path}, error: {e}",));
-        cmd.stderr_to_stdout().stdout_file(log_file)
-    };
-
-    Box::new(with_logs)
+// The description will be logged at the info log level. The command will be modified and converted to
+// a tokio::process::Command, redirecting stdout and stderr.
+//
+// If the log is None, stdout and stderr will be redirected to the caller's stdout.
+// If it contains a path, an attempt will be made to use it to redirect the command's
+// stdout and stderr. If, for any reason, the log file cannot be opened or created,
+// redirection will default back to the caller's stdout.
+pub trait WithLogs {
+    fn with_logs(self, description: &str, log_path: &Option<PathBuf>) -> TokioCommand;
+    fn spawn_with_logs(
+        self,
+        description: &str,
+        log_path: &Option<PathBuf>,
+    ) -> anyhow::Result<Child>;
+    fn run_with_logs(
+        self,
+        description: &str,
+        log_path: &Option<PathBuf>,
+    ) -> JoinHandle<anyhow::Result<()>>;
+}
+
+pub trait ProcessGroupId {
+    fn process_group(self, pgid: i32) -> StdCommand;
+}
+
+impl WithLogs for StdCommand {
+    fn with_logs(self, description: &str, log_path: &Option<PathBuf>) -> TokioCommand {
+        tracing::info!("{description}");
+
+        let (stdout, stderr) = log_path
+            .as_ref()
+            .and_then(|log_path| {
+                match std::fs::File::options()
+                    .append(true)
+                    .create(true)
+                    .open(log_path.clone())
+                {
+                    // If log file exists then use it
+                    Ok(mut log_out_file) => {
+                        let Ok(log_err_file) = log_out_file.try_clone() else {
+                            return Some((Stdio::inherit(), Stdio::inherit()));
+                        };
+
+                        let _ = log_out_file
+                            .write(format!("{}\n", description).as_bytes())
+                            .map_err(|e| {
+                                warn!("Error writing into {}, error: {e}", log_path.display())
+                            });
+                        let _ = log_out_file.flush().map_err(|e| {
+                            warn!("Error writing into {}, error: {e}", log_path.display())
+                        });
+                        Some((Stdio::from(log_out_file), Stdio::from(log_err_file)))
+                    }
+                    // If log file does not exist then use inherited stdout and stderr
+                    Err(_) => Some((Stdio::inherit(), Stdio::inherit())),
+                }
+            })
+            // If log file is not specified use inherited stdout and stderr
+            .unwrap_or((Stdio::inherit(), Stdio::inherit()));
+
+        let mut command = TokioCommand::from(self);
+        command.stderr(stderr).stdout(stdout);
+        command
+    }
+
+    fn spawn_with_logs(
+        self,
+        description: &str,
+        log_path: &Option<PathBuf>,
+    ) -> anyhow::Result<tokio::process::Child> {
+        self.with_logs(description, log_path)
+            .kill_on_drop(true)
+            .spawn()
+            .map_err(|e| e.into())
+    }
+
+    fn run_with_logs(
+        self,
+        description: &str,
+        log_path: &Option<PathBuf>,
+    ) -> tokio::task::JoinHandle<anyhow::Result<()>> {
+        let description = String::from(description);
+        let log_path = log_path.clone();
+        tokio::task::spawn(async move {
+            let exit_status: anyhow::Result<std::process::ExitStatus> = self
+                .spawn_with_logs(&description, &log_path)?
+                .wait()
+                .await
+                .map_err(|e| e.into());
+            match exit_status?.code() {
+                Some(code) if code != 0 => Err(anyhow::anyhow!(
+                    "{description}, exit with status code: {code}",
+                )),
+                _ => Ok(()),
+            }
+        })
+    }
+}
+
+impl<'a> WithLogs for xshell::Cmd<'a> {
+    fn with_logs(self, description: &str, log_path: &Option<PathBuf>) -> TokioCommand {
+        StdCommand::from(self).with_logs(description, log_path)
+    }
+
+    fn spawn_with_logs(
+        self,
+        description: &str,
+        log_path: &Option<PathBuf>,
+    ) -> anyhow::Result<tokio::process::Child> {
+        StdCommand::from(self).spawn_with_logs(description, log_path)
+    }
+
+    fn run_with_logs(
+        self,
+        description: &str,
+        log_path: &Option<PathBuf>,
+    ) -> tokio::task::JoinHandle<anyhow::Result<()>> {
+        StdCommand::from(self).run_with_logs(description, log_path)
+    }
+}
+
+impl<'a> ProcessGroupId for xshell::Cmd<'a> {
+    fn process_group(self, pgid: i32) -> StdCommand {
+        let mut command = StdCommand::from(self);
+        command.process_group(pgid);
+        command
+    }
 }