diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index 1c652870..c8853a6b 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -10143,28 +10143,25 @@ pub async fn run_cron_job( } }; let job_id = openfang_types::scheduler::CronJobId(uuid); - let job = match state.kernel.cron_scheduler.get_job(job_id) { - Some(j) => j, - None => { + + // Atomically check existence + enabled + reserve next_run in one lock hold. + let job = match state.kernel.cron_scheduler.try_claim_for_run(job_id) { + Ok(j) => j, + Err(openfang_kernel::cron::ClaimError::NotFound) => { return ( StatusCode::NOT_FOUND, Json(serde_json::json!({"status": "error", "error": "Job not found"})), ); } + Err(openfang_kernel::cron::ClaimError::Disabled) => { + return ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({"status": "error", "error": "Job is disabled"})), + ); + } }; - if !job.enabled { - return ( - StatusCode::BAD_REQUEST, - Json(serde_json::json!({"status": "error", "error": "Job is disabled"})), - ); - } - - // Pre-advance next_run so the background scheduler tick won't also fire - // this job while the manual run is in progress. - state.kernel.cron_scheduler.reserve_run(job_id); // Spawn execution in the background so we don't block the HTTP response. - // `job` is already a clone from `get_job()`, so move it directly. let kernel = Arc::clone(&state.kernel); let job_name = job.name.clone(); tokio::spawn(async move { diff --git a/crates/openfang-kernel/src/cron.rs b/crates/openfang-kernel/src/cron.rs index 723ba989..07c4fedf 100644 --- a/crates/openfang-kernel/src/cron.rs +++ b/crates/openfang-kernel/src/cron.rs @@ -20,6 +20,15 @@ use tracing::{debug, info, warn}; /// Maximum consecutive errors before a job is auto-disabled. const MAX_CONSECUTIVE_ERRORS: u32 = 5; +/// Error returned by [`CronScheduler::try_claim_for_run`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ClaimError { + /// Job ID does not exist. + NotFound, + /// Job exists but is disabled. + Disabled, +} + // --------------------------------------------------------------------------- // JobMeta — extra runtime state not stored in CronJob itself // --------------------------------------------------------------------------- @@ -306,21 +315,29 @@ impl CronScheduler { due } - /// Pre-advance a job's `next_run` so the background scheduler won't also - /// fire it while a manual (on-demand) run is in progress. + /// Atomically look up a job, verify it is enabled, pre-advance `next_run` + /// if overdue, and return a snapshot of the job for execution. /// - /// Only advances `next_run` when the job is already due (`next_run <= now`), - /// matching the same guard used by `due_jobs()`. This avoids silently - /// skipping an imminent scheduled run when the user triggers a manual run - /// on a job that isn't due yet. + /// This combines the existence check, enabled guard, and scheduler + /// reservation into a single `DashMap::get_mut` hold so no concurrent + /// request can disable/delete the job between the check and the claim. /// - /// Call this **before** spawning the manual execution task to close the race - /// window between the API handler and the next scheduler tick. - pub fn reserve_run(&self, id: CronJobId) { - if let Some(mut meta) = self.jobs.get_mut(&id) { - let now = Utc::now(); - if meta.job.next_run.map(|t| t <= now).unwrap_or(false) { - meta.job.next_run = Some(compute_next_run_after(&meta.job.schedule, now)); + /// `next_run` is only advanced when the job is already due + /// (`next_run <= now`), matching `due_jobs()`, so triggering a manual run + /// on a not-yet-due job won't skip the upcoming scheduled fire. + pub fn try_claim_for_run(&self, id: CronJobId) -> Result { + match self.jobs.get_mut(&id) { + None => Err(ClaimError::NotFound), + Some(mut entry) => { + let meta = entry.value_mut(); + if !meta.job.enabled { + return Err(ClaimError::Disabled); + } + let now = Utc::now(); + if meta.job.next_run.map(|t| t <= now).unwrap_or(false) { + meta.job.next_run = Some(compute_next_run_after(&meta.job.schedule, now)); + } + Ok(meta.job.clone()) } } } @@ -1227,14 +1244,30 @@ mod tests { } } - // -- reserve_run --------------------------------------------------------- + // -- try_claim_for_run --------------------------------------------------- #[test] - fn reserve_run_skips_not_yet_due_job() { + fn try_claim_not_found() { + let (sched, _tmp) = make_scheduler(100); + let id = CronJobId::new(); + assert!(matches!(sched.try_claim_for_run(id), Err(ClaimError::NotFound))); + } + + #[test] + fn try_claim_disabled() { + let (sched, _tmp) = make_scheduler(100); + let agent = AgentId::new(); + let mut job = make_job(agent); + job.enabled = false; + let id = sched.add_job(job, false).unwrap(); + assert!(matches!(sched.try_claim_for_run(id), Err(ClaimError::Disabled))); + } + + #[test] + fn try_claim_skips_not_yet_due_job() { let (sched, _tmp) = make_scheduler(100); let agent = AgentId::new(); let mut job = make_job(agent); - // Schedule is Every 3600s; next_run will be ~1 hour from now after add. job.schedule = CronSchedule::Every { every_secs: 3600 }; let id = sched.add_job(job, false).unwrap(); @@ -1242,14 +1275,15 @@ mod tests { assert!(original_next_run.is_some()); // Manual trigger on a not-yet-due job should NOT move next_run. - sched.reserve_run(id); + let claimed = sched.try_claim_for_run(id).unwrap(); + assert_eq!(claimed.id, id); let after = sched.get_job(id).unwrap().next_run; - assert_eq!(original_next_run, after, "reserve_run must not move next_run for a job that is not yet due"); + assert_eq!(original_next_run, after, "try_claim must not move next_run for a job that is not yet due"); } #[test] - fn reserve_run_advances_overdue_job() { + fn try_claim_advances_overdue_job() { let (sched, _tmp) = make_scheduler(100); let agent = AgentId::new(); let mut job = make_job(agent); @@ -1264,9 +1298,10 @@ mod tests { let before = sched.get_job(id).unwrap().next_run.unwrap(); assert!(before < Utc::now(), "precondition: job should be overdue"); - sched.reserve_run(id); + let claimed = sched.try_claim_for_run(id).unwrap(); + assert_eq!(claimed.id, id); let after = sched.get_job(id).unwrap().next_run.unwrap(); - assert!(after > Utc::now(), "reserve_run should advance next_run past now for overdue jobs"); + assert!(after > Utc::now(), "try_claim should advance next_run past now for overdue jobs"); } }