feat: add edit schedule functionality (#2700)

This commit is contained in:
Max Novich
2025-05-28 13:58:22 -07:00
committed by GitHub
parent ea6a7a7847
commit 132e0b7fca
10 changed files with 969 additions and 55 deletions
+132
View File
@@ -577,6 +577,138 @@ impl Scheduler {
None => Err(SchedulerError::JobNotFound(sched_id.to_string())),
}
}
pub async fn update_schedule(
&self,
sched_id: &str,
new_cron: String,
) -> Result<(), SchedulerError> {
let mut jobs_guard = self.jobs.lock().await;
match jobs_guard.get_mut(sched_id) {
Some((job_uuid, job_def)) => {
if job_def.currently_running {
return Err(SchedulerError::AnyhowError(anyhow!(
"Cannot edit schedule '{}' while it's currently running",
sched_id
)));
}
if new_cron == job_def.cron {
// No change needed
return Ok(());
}
// Remove the old job from the scheduler
self.internal_scheduler
.remove(job_uuid)
.await
.map_err(|e| SchedulerError::SchedulerInternalError(e.to_string()))?;
// Create new job with updated cron
let job_for_task = job_def.clone();
let jobs_arc_for_task = self.jobs.clone();
let storage_path_for_task = self.storage_path.clone();
let cron_task = Job::new_async(&new_cron, move |_uuid, _l| {
let task_job_id = job_for_task.id.clone();
let current_jobs_arc = jobs_arc_for_task.clone();
let local_storage_path = storage_path_for_task.clone();
let job_to_execute = job_for_task.clone();
Box::pin(async move {
// Check if the job is paused before executing
let should_execute = {
let jobs_map_guard = current_jobs_arc.lock().await;
if let Some((_, current_job_in_map)) = jobs_map_guard.get(&task_job_id)
{
!current_job_in_map.paused
} else {
false
}
};
if !should_execute {
tracing::info!("Skipping execution of paused job '{}'", &task_job_id);
return;
}
let current_time = Utc::now();
let mut needs_persist = false;
{
let mut jobs_map_guard = current_jobs_arc.lock().await;
if let Some((_, current_job_in_map)) =
jobs_map_guard.get_mut(&task_job_id)
{
current_job_in_map.last_run = Some(current_time);
current_job_in_map.currently_running = true;
needs_persist = true;
}
}
if needs_persist {
if let Err(e) =
persist_jobs_from_arc(&local_storage_path, &current_jobs_arc).await
{
tracing::error!(
"Failed to persist last_run update for job {}: {}",
&task_job_id,
e
);
}
}
let result = run_scheduled_job_internal(job_to_execute, None).await;
// Update the job status after execution
{
let mut jobs_map_guard = current_jobs_arc.lock().await;
if let Some((_, current_job_in_map)) =
jobs_map_guard.get_mut(&task_job_id)
{
current_job_in_map.currently_running = false;
needs_persist = true;
}
}
if needs_persist {
if let Err(e) =
persist_jobs_from_arc(&local_storage_path, &current_jobs_arc).await
{
tracing::error!(
"Failed to persist running status update for job {}: {}",
&task_job_id,
e
);
}
}
if let Err(e) = result {
tracing::error!(
"Scheduled job '{}' execution failed: {}",
&e.job_id,
e.error
);
}
})
})
.map_err(|e| SchedulerError::CronParseError(e.to_string()))?;
let new_job_uuid = self
.internal_scheduler
.add(cron_task)
.await
.map_err(|e| SchedulerError::SchedulerInternalError(e.to_string()))?;
// Update the job UUID and cron expression
*job_uuid = new_job_uuid;
job_def.cron = new_cron;
self.persist_jobs_to_storage_with_guard(&jobs_guard).await?;
Ok(())
}
None => Err(SchedulerError::JobNotFound(sched_id.to_string())),
}
}
}
#[derive(Debug)]