scheduler: add per-job concurrency control

Add concurrency strategies to control overlapping job executions:
- parallel: allow concurrent runs
- wait: wait for previous run to complete
- skip: skip if previous run still active (default)
- replace: kill previous run, start new

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: aster-void

SPEC.md

@@ -48,8 +48,18 @@ jobs:
       cron: "<cron-expr>"      # Required: 5-field cron expression
     run: <string>              # Required: shell command to execute
     timeout: <duration>        # Optional: default "10s"
+    concurrency: <strategy>    # Optional: default "skip"
 ```
 
+### Concurrency Strategies
+
+| Strategy | Behavior |
+|----------|----------|
+| `parallel` | Allow concurrent runs |
+| `wait` | Wait for previous run to complete |
+| `skip` | Skip if previous run still active (default) |
+| `replace` | Kill previous run, start new |
+
 ### Example
 
 ```yaml
@@ -60,11 +70,13 @@ jobs:
       cron: "0 * * * *"
     run: cargo build --release
     timeout: 5m
+    concurrency: wait
 
   health-check:
     schedule:
       cron: "*/5 * * * *"
     run: curl -f http://localhost/health
+    concurrency: skip
 ```
 
 ### Cron Expression

src/config.rs

@@ -5,6 +5,16 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use std::time::Duration;
 
+#[derive(Debug, Clone, Default, Deserialize, PartialEq)]
+#[serde(rename_all = "lowercase")]
+pub enum Concurrency {
+    Parallel,
+    Wait,
+    #[default]
+    Skip,
+    Replace,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct Config {
     pub jobs: HashMap<String, JobConfig>,
@@ -17,6 +27,8 @@ pub struct JobConfig {
     pub run: String,
     #[serde(default = "default_timeout")]
     pub timeout: String,
+    #[serde(default)]
+    pub concurrency: Concurrency,
 }
 
 #[derive(Debug, Deserialize)]
@@ -35,6 +47,7 @@ pub struct Job {
     pub schedule: Schedule,
     pub command: String,
     pub timeout: Duration,
+    pub concurrency: Concurrency,
 }
 
 pub fn parse_config(content: &str) -> Result<Vec<Job>> {
@@ -60,6 +73,7 @@ pub fn parse_config(content: &str) -> Result<Vec<Job>> {
                 schedule,
                 command: job.run,
                 timeout,
+                concurrency: job.concurrency,
             })
         })
         .collect()
@@ -153,4 +167,51 @@ jobs:
         assert_eq!(parse_duration("5m").unwrap(), Duration::from_secs(300));
         assert_eq!(parse_duration("1h").unwrap(), Duration::from_secs(3600));
     }
+
+    #[test]
+    fn parse_concurrency_default() {
+        let yaml = r#"
+jobs:
+  test:
+    schedule:
+      cron: "* * * * *"
+    run: echo test
+"#;
+        let jobs = parse_config(yaml).unwrap();
+        assert_eq!(jobs[0].concurrency, Concurrency::Skip);
+    }
+
+    #[test]
+    fn parse_concurrency_all_variants() {
+        let yaml = r#"
+jobs:
+  parallel_job:
+    schedule:
+      cron: "* * * * *"
+    run: echo 1
+    concurrency: parallel
+  wait_job:
+    schedule:
+      cron: "* * * * *"
+    run: echo 2
+    concurrency: wait
+  skip_job:
+    schedule:
+      cron: "* * * * *"
+    run: echo 3
+    concurrency: skip
+  replace_job:
+    schedule:
+      cron: "* * * * *"
+    run: echo 4
+    concurrency: replace
+"#;
+        let jobs = parse_config(yaml).unwrap();
+        let find = |id: &str| jobs.iter().find(|j| j.id == id).unwrap();
+
+        assert_eq!(find("parallel_job").concurrency, Concurrency::Parallel);
+        assert_eq!(find("wait_job").concurrency, Concurrency::Wait);
+        assert_eq!(find("skip_job").concurrency, Concurrency::Skip);
+        assert_eq!(find("replace_job").concurrency, Concurrency::Replace);
+    }
 }

src/scheduler.rs

@@ -1,25 +1,29 @@
-use crate::config::Job;
+use crate::config::{Concurrency, Job};
 use crate::git;
 use anyhow::Result;
 use chrono::Utc;
+use std::collections::HashMap;
 use std::path::PathBuf;
 use std::process::Command;
+use std::sync::Arc;
 use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::task::JoinHandle;
 use tokio::time::sleep;
 
 pub async fn run_scheduler(jobs: Vec<Job>, sot_path: PathBuf) -> Result<()> {
+    let running: Arc<Mutex<HashMap<String, Vec<JoinHandle<()>>>>> =
+        Arc::new(Mutex::new(HashMap::new()));
+
     loop {
         let now = Utc::now();
 
         for job in &jobs {
             if let Some(next) = job.schedule.upcoming(Utc).next() {
                 let until_next = (next - now).num_milliseconds();
                 if until_next <= 1000 && until_next >= 0 {
-                    let job = job.clone();
                     let work_dir = git::get_job_dir(&sot_path, &job.id);
-                    tokio::spawn(async move {
-                        execute_job(&job, &work_dir).await;
-                    });
+                    handle_job_trigger(job, work_dir, &running).await;
                 }
             }
         }
@@ -28,6 +32,76 @@ pub async fn run_scheduler(jobs: Vec<Job>, sot_path: PathBuf) -> Result<()> {
     }
 }
 
+async fn handle_job_trigger(
+    job: &Job,
+    work_dir: PathBuf,
+    running: &Arc<Mutex<HashMap<String, Vec<JoinHandle<()>>>>>,
+) {
+    let tag = format!("[job:{}]", job.id);
+    let mut map = running.lock().await;
+
+    // Clean up finished jobs
+    if let Some(handles) = map.get_mut(&job.id) {
+        handles.retain(|h| !h.is_finished());
+        if handles.is_empty() {
+            map.remove(&job.id);
+        }
+    }
+
+    let running_count = map.get(&job.id).map(|v| v.len()).unwrap_or(0);
+
+    match job.concurrency {
+        Concurrency::Parallel => {
+            spawn_job(job, work_dir, &mut map);
+        }
+        Concurrency::Wait => {
+            if running_count > 0 {
+                println!("{} Waiting for {} previous run(s) to complete", tag, running_count);
+                let handles = map.remove(&job.id).unwrap();
+                drop(map); // Release lock while waiting
+                for handle in handles {
+                    let _ = handle.await;
+                }
+                let mut map = running.lock().await;
+                spawn_job(job, work_dir, &mut map);
+            } else {
+                spawn_job(job, work_dir, &mut map);
+            }
+        }
+        Concurrency::Skip => {
+            if running_count > 0 {
+                println!("{} Skipped ({} instance(s) still active)", tag, running_count);
+            } else {
+                spawn_job(job, work_dir, &mut map);
+            }
+        }
+        Concurrency::Replace => {
+            if running_count > 0 {
+                println!("{} Replacing {} previous run(s)", tag, running_count);
+                let handles = map.remove(&job.id).unwrap();
+                for handle in handles {
+                    handle.abort();
+                }
+            }
+            spawn_job(job, work_dir, &mut map);
+        }
+    }
+}
+
+fn spawn_job(
+    job: &Job,
+    work_dir: PathBuf,
+    map: &mut HashMap<String, Vec<JoinHandle<()>>>,
+) {
+    let job_id = job.id.clone();
+    let job = job.clone();
+    let handle = tokio::spawn(async move {
+        execute_job(&job, &work_dir).await;
+    });
+
+    map.entry(job_id).or_default().push(handle);
+}
+
 async fn execute_job(job: &Job, work_dir: &PathBuf) {
     let tag = format!("[job:{}]", job.id);
 
@@ -101,6 +175,7 @@ mod tests {
             schedule: Schedule::from_str("* * * * * *").unwrap(),
             command: cmd.to_string(),
             timeout: Duration::from_secs(timeout_secs),
+            concurrency: Concurrency::Skip,
         }
     }