Avoid races on cron

Author: dcvz

src/cron/mod.rs

@@ -60,7 +60,13 @@ pub struct CronService<C: Clock> {
 impl<C: Clock> CronService<C> {
     /// Create a new cron service.
     pub fn new(clock: C, config: CronConfig) -> Result<Self> {
-        let store = CronStore::load()?;
+        let mut store = CronStore::load()?;
+
+        let recovered = store.recover_stuck_jobs(clock.now_millis());
+        if recovered > 0 {
+            info!("Recovered {} stuck cron job(s) from previous run", recovered);
+            let _ = store.save();
+        }
 
         Ok(Self {
             clock,
@@ -93,12 +99,12 @@ impl<C: Clock> CronService<C> {
                         break;
                     }
                     _ = clock.sleep(tick_interval) => {
-                        // Reload store from disk to pick up external changes
-                        // (e.g., agent modifying cron.json directly)
+                        // Merge disk changes (e.g., agent modifying cron.json)
+                        // while preserving in-memory state for running jobs
                         {
                             let mut store_guard = store.lock().await;
                             match CronStore::load() {
-                                Ok(fresh) => *store_guard = fresh,
+                                Ok(fresh) => store_guard.merge_from_disk(fresh),
                                 Err(e) => warn!("Failed to reload cron store: {}", e),
                             }
                         }

src/cron/store.rs

@@ -243,6 +243,53 @@ impl CronStore {
     pub fn get_enabled_jobs(&self) -> Vec<&CronJob> {
         self.jobs.values().filter(|j| j.enabled).collect()
     }
+
+    /// Reset any jobs stuck in Running state (e.g., after a crash).
+    /// Recalculates next_run_at so they get scheduled again.
+    pub fn recover_stuck_jobs(&mut self, now_ms: u64) -> usize {
+        let stuck_ids: Vec<JobId> = self
+            .jobs
+            .values()
+            .filter(|j| j.state.last_status == JobStatus::Running)
+            .map(|j| j.id.clone())
+            .collect();
+
+        let count = stuck_ids.len();
+        for id in &stuck_ids {
+            if let Some(job) = self.jobs.get_mut(id) {
+                job.state.last_status = JobStatus::Success;
+                job.update_next_run(now_ms);
+            }
+        }
+
+        count
+    }
+
+    /// Merge disk state into the current store, preserving in-flight job states.
+    /// Jobs currently marked as Running in memory keep their in-memory state
+    /// to avoid losing completion updates from concurrent tasks.
+    pub fn merge_from_disk(&mut self, disk: CronStore) {
+        let running_ids: std::collections::HashSet<String> = self
+            .jobs
+            .values()
+            .filter(|j| j.state.last_status == JobStatus::Running)
+            .map(|j| j.id.clone())
+            .collect();
+
+        let disk_ids: std::collections::HashSet<String> =
+            disk.jobs.keys().cloned().collect();
+
+        for (id, disk_job) in disk.jobs {
+            if running_ids.contains(&id) {
+                continue;
+            }
+            self.jobs.insert(id, disk_job);
+        }
+
+        self.jobs.retain(|id, _| {
+            disk_ids.contains(id) || running_ids.contains(id)
+        });
+    }
 }
 
 /// Generate a unique job ID.