feat: refactor the code structure of start and run, allowing for parallel starting daemons (#56)

* feat: refactor the code structure of `start` and `run`, allowing for parallel starting daemons

1. Use tokio to run daemons in parallel, which avoids waiting 3 secs (by default) for EACH daemon.

2. Now that pitchfork will always exit with code 1, regardless of what exit code the daemons give.
And we have already printed messages like `daemon xxx exited with code xxx`.

3. Move most of the message output to ipc client, to avoid duplicating them in `start` and `run` command.

* fix: pass exit code on retrying correctly

Author: gaojunran

README.md

@@ -88,16 +88,12 @@ run = "npm run server:api"
 run = "npm run server:docs"
 ```
 
-Start all daemons:
+Start all daemons or mutiple daemons **in parallel**:
 
 ```sh-session
 $ pitchfork start --all
-```
-
-Or start individual ones:
 
-```sh-session
-$ pitchfork start redis
+$ pitchfork start redis api
 ```
 
 ### Shell hook (auto start/stop)
@@ -144,7 +140,7 @@ ready.http = { url = "http://localhost:5432" }
 [daemons.redis]
 run = "redis-server --port 6379"
 auto = ["start", "stop"]
-ready.delay = "2s"
+ready.delay = 2
 
 [daemons.api]
 run = "npm run dev:api"

docs/getting-started.md

@@ -18,8 +18,8 @@ This project is experimental. It works in basic situations but you'll undoubtedl
 - Auto start daemons when entering a project directory - then auto stop when leaving
 - Restart daemons on failure
 - Cron jobs
-- 🚧 Global configuration
-- 🚧 Automatically start daemons on boot
+- Global configuration
+- Automatically start daemons on boot
 
 ## Workflows
 

src/cli/run.rs

@@ -26,15 +26,13 @@ pub struct Run {
 
 impl Run {
     pub async fn run(&self) -> Result<()> {
-        info!("Running one-off daemon");
         if self.run.is_empty() {
             bail!("No command provided");
         }
 
-        let start_time = chrono::Local::now();
         let ipc = IpcClient::connect(true).await?;
 
-        let (started, exit_code) = ipc
+        let (_started, exit_code) = ipc
             .run(RunOptions {
                 id: self.id.clone(),
                 cmd: self.run.clone(),
@@ -52,20 +50,8 @@ impl Run {
             })
             .await?;
 
-        if !started.is_empty() {
-            info!("started {}", started.join(", "));
-        }
-
-        if let Some(code) = exit_code {
-            error!("Daemon '{}' failed with exit code {}", self.id, code);
-
-            // Print logs from the time we started
-            if let Err(e) = crate::cli::logs::print_logs_for_time_range(&self.id, start_time, None)
-            {
-                error!("Failed to print logs: {}", e);
-            }
-
-            std::process::exit(code);
+        if exit_code.is_some() {
+            std::process::exit(1);
         }
         Ok(())
     }

src/cli/start.rs

@@ -2,8 +2,8 @@ use crate::daemon::RunOptions;
 use crate::ipc::client::IpcClient;
 use crate::pitchfork_toml::PitchforkToml;
 use crate::Result;
-use miette::{ensure, IntoDiagnostic};
-use std::collections::HashSet;
+use miette::ensure;
+use std::sync::Arc;
 
 /// Starts a daemon from a pitchfork.toml file
 #[derive(Debug, clap::Args)]
@@ -34,86 +34,131 @@ impl Start {
             "At least one daemon ID must be provided"
         );
         let pt = PitchforkToml::all_merged();
-        let ipc = IpcClient::connect(true).await?;
+        let ipc = Arc::new(IpcClient::connect(true).await?);
         let disabled_daemons = ipc.get_disabled_daemons().await?;
-        let active_daemons: HashSet<String> = ipc
-            .active_daemons()
-            .await?
-            .into_iter()
-            .map(|d| d.id)
-            .collect();
         let ids = if self.all {
             pt.daemons.keys().cloned().collect()
         } else {
             self.id.clone()
         };
-        let mut any_failed = false;
-        let mut last_exit_code = 0;
+        // launch all tasks concurrently
+        let mut tasks = Vec::new();
 
-        for id in &ids {
-            if disabled_daemons.contains(id) {
+        for id in ids {
+            if disabled_daemons.contains(&id) {
                 warn!("Daemon {} is disabled", id);
                 continue;
             }
-            if !self.force && active_daemons.contains(id) {
-                warn!("Daemon {} is already running", id);
-                continue;
-            }
-            let daemon = pt.daemons.get(id);
-            if let Some(daemon) = daemon {
-                info!("Starting daemon {}", id);
-                let start_time = chrono::Local::now();
-                let cmd = shell_words::split(&daemon.run).into_diagnostic()?;
-                let (started, exit_code) = ipc
+
+            let daemon_data = match pt.daemons.get(&id) {
+                Some(d) => {
+                    let run = d.run.clone();
+                    let auto_stop = d
+                        .auto
+                        .contains(&crate::pitchfork_toml::PitchforkTomlAuto::Stop);
+                    let dir = d
+                        .path
+                        .as_ref()
+                        .and_then(|p| p.parent())
+                        .map(|p| p.to_path_buf())
+                        .unwrap_or_default();
+                    let cron_schedule = d.cron.as_ref().map(|c| c.schedule.clone());
+                    let cron_retrigger = d.cron.as_ref().map(|c| c.retrigger);
+                    let retry = d.retry;
+                    let ready_delay = d.ready_delay;
+                    let ready_output = d.ready_output.clone();
+
+                    (
+                        run,
+                        auto_stop,
+                        dir,
+                        cron_schedule,
+                        cron_retrigger,
+                        retry,
+                        ready_delay,
+                        ready_output,
+                    )
+                }
+                None => {
+                    warn!("Daemon {} not found", id);
+                    continue;
+                }
+            };
+
+            let (
+                run,
+                auto_stop,
+                dir,
+                cron_schedule,
+                cron_retrigger,
+                retry,
+                ready_delay,
+                ready_output,
+            ) = daemon_data;
+
+            let ipc_clone = ipc.clone();
+            let shell_pid = self.shell_pid;
+            let force = self.force;
+            let delay = self.delay;
+            let output = self.output.clone();
+
+            let task = tokio::spawn(async move {
+                let cmd = match shell_words::split(&run) {
+                    Ok(c) => c,
+                    Err(e) => {
+                        error!("Failed to parse command for daemon {}: {}", id, e);
+                        return Some(1);
+                    }
+                };
+
+                match ipc_clone
                     .run(RunOptions {
                         id: id.clone(),
                         cmd,
-                        shell_pid: self.shell_pid,
-                        force: self.force,
-                        autostop: daemon
-                            .auto
-                            .contains(&crate::pitchfork_toml::PitchforkTomlAuto::Stop),
-                        dir: daemon
-                            .path
-                            .as_ref()
-                            .unwrap()
-                            .parent()
-                            .map(|p| p.to_path_buf())
-                            .unwrap_or_default(),
-                        cron_schedule: daemon.cron.as_ref().map(|c| c.schedule.clone()),
-                        cron_retrigger: daemon.cron.as_ref().map(|c| c.retrigger),
-                        retry: daemon.retry,
+                        shell_pid,
+                        force,
+                        autostop: auto_stop,
+                        dir,
+                        cron_schedule,
+                        cron_retrigger,
+                        retry,
                         retry_count: 0,
-                        ready_delay: self.delay.or(daemon.ready_delay).or(Some(3)),
-                        ready_output: self.output.clone().or(daemon.ready_output.clone()),
+                        ready_delay: delay.or(ready_delay).or(Some(3)),
+                        ready_output: output.or(ready_output),
                         wait_ready: true,
                     })
-                    .await?;
-                if !started.is_empty() {
-                    info!("started {}", started.join(", "));
+                    .await
+                {
+                    Ok((_started, exit_code)) => exit_code,
+                    Err(e) => {
+                        error!("Failed to start daemon {}: {}", id, e);
+                        Some(1)
+                    }
                 }
-                if let Some(code) = exit_code {
-                    any_failed = true;
-                    last_exit_code = code;
-                    error!("daemon {} failed with exit code {}", id, code);
+            });
+
+            tasks.push(task);
+        }
 
-                    // Print logs from the time we started this specific daemon
-                    if let Err(e) =
-                        crate::cli::logs::print_logs_for_time_range(id, start_time, None)
-                    {
-                        error!("Failed to print logs: {}", e);
+        // wait for all tasks to complete
+        let mut any_failed = false;
+
+        for task in tasks {
+            match task.await {
+                Ok(exit_code) => {
+                    if exit_code.is_some() {
+                        any_failed = true;
                     }
                 }
-            } else {
-                warn!("Daemon {} not found", id);
+                Err(e) => {
+                    error!("Task panicked: {}", e);
+                    any_failed = true;
+                }
             }
         }
 
         if any_failed {
-            if last_exit_code != 0 {
-                error!("Process exited with code {}", last_exit_code);
-            }
-            std::process::exit(last_exit_code);
+            std::process::exit(1);
         }
         Ok(())
     }

src/ipc/client.rs

@@ -135,25 +135,45 @@ impl IpcClient {
     }
 
     pub async fn run(&self, opts: RunOptions) -> Result<(Vec<String>, Option<i32>)> {
-        debug!("starting daemon {}", opts.id);
+        info!("starting daemon {}", opts.id);
+        let start_time = chrono::Local::now();
         let rsp = self.request(IpcRequest::Run(opts.clone())).await?;
         let mut started_daemons = vec![];
         let mut exit_code = None;
         match rsp {
             IpcResponse::DaemonStart { daemon } => {
-                started_daemons.push(daemon.id);
+                started_daemons.push(daemon.id.clone());
+                info!("started {}", daemon.id);
             }
             IpcResponse::DaemonReady { daemon } => {
-                started_daemons.push(daemon.id);
+                started_daemons.push(daemon.id.clone());
+                info!("started {}", daemon.id);
             }
             IpcResponse::DaemonFailedWithCode { exit_code: code } => {
-                exit_code = Some(code.unwrap_or(1));
+                let code = code.unwrap_or(1);
+                exit_code = Some(code);
+                error!("daemon {} failed with exit code {}", opts.id, code);
+
+                // Print logs from the time we started this specific daemon
+                if let Err(e) =
+                    crate::cli::logs::print_logs_for_time_range(&opts.id, start_time, None)
+                {
+                    error!("Failed to print logs: {}", e);
+                }
             }
             IpcResponse::DaemonAlreadyRunning => {
                 warn!("daemon {} already running", opts.id);
             }
             IpcResponse::DaemonFailed { error } => {
-                bail!("failed to start daemon {}: {error}", opts.id);
+                error!("Failed to start daemon {}: {}", opts.id, error);
+                exit_code = Some(1);
+
+                // Print logs from the time we started this specific daemon
+                if let Err(e) =
+                    crate::cli::logs::print_logs_for_time_range(&opts.id, start_time, None)
+                {
+                    error!("Failed to print logs: {}", e);
+                }
             }
             rsp => unreachable!("unexpected response: {rsp:?}"),
         }

src/supervisor.rs

@@ -486,12 +486,25 @@ impl Supervisor {
                         debug!("daemon {id} process exited, exit_status: {:?}", exit_status);
                         if !ready_notified {
                             if let Some(tx) = ready_tx.take() {
-                                let exit_code = exit_status.as_ref().and_then(|r| r.as_ref().ok().and_then(|s| s.code()));
-                                debug!("daemon {id} not ready yet, sending failure notification with exit_code: {:?}", exit_code);
-                                let _ = tx.send(Err(exit_code));
+                                // Check if process exited successfully
+                                let is_success = exit_status.as_ref()
+                                    .and_then(|r| r.as_ref().ok())
+                                    .map(|s| s.success())
+                                    .unwrap_or(false);
+
+                                if is_success {
+                                    debug!("daemon {id} exited successfully before ready check, sending success notification");
+                                    let _ = tx.send(Ok(()));
+                                } else {
+                                    let exit_code = exit_status.as_ref()
+                                        .and_then(|r| r.as_ref().ok())
+                                        .and_then(|s| s.code());
+                                    debug!("daemon {id} exited with failure before ready check, sending failure notification with exit_code: {:?}", exit_code);
+                                    let _ = tx.send(Err(exit_code));
+                                }
                             }
                         } else {
-                            debug!("daemon {id} was already marked ready, not sending failure notification");
+                            debug!("daemon {id} was already marked ready, not sending notification");
                         }
                         break;
                     }

tests/test_e2e.rs

@@ -624,10 +624,6 @@ ready_output = "NEVER_APPEARS"
 
     // When ready_output is set but never matches, it blocks until daemon exits
     // slowly_output.ts outputs 5 times with 1s interval, so runs for ~5 seconds
-    assert!(
-        output.status.success(),
-        "Start should succeed when daemon exits cleanly even if pattern never matches"
-    );
     assert!(
         elapsed >= Duration::from_secs(3),
         "Should block until daemon exits (~3s), took {:?}",