apollo/
cron_scheduler.rs

1//! SurrealDB-backed Cron Scheduler — persistent scheduled tasks.
2//! Stores jobs in SurrealDB, ticks every 60s, spawns agent sessions for due jobs.
3
4use crate::memory::surreal::SurrealMemory;
5use serde::{Deserialize, Serialize};
6use std::str::FromStr;
7use std::sync::Arc;
8
9/// A cron job stored in SurrealDB.
10#[derive(Debug, Clone, Serialize, Deserialize)]
11pub struct CronJob {
12    pub id: Option<String>,
13    pub name: String,
14    pub schedule: String,
15    #[serde(default)]
16    pub one_shot: bool,
17    pub task: String,
18    pub channel: String,
19    #[serde(default)]
20    pub chat_id: String,
21    pub model: String,
22    pub enabled: bool,
23    pub last_run: Option<String>,
24    pub next_run: Option<String>,
25    #[serde(default = "default_job_status")]
26    pub status: String,
27    #[serde(default)]
28    pub retry_count: u32,
29    #[serde(default = "default_max_retries")]
30    pub max_retries: u32,
31    #[serde(default)]
32    pub last_error: Option<String>,
33    #[serde(default)]
34    pub lease_until: Option<String>,
35    #[serde(default)]
36    pub run_token: Option<String>,
37}
38
39fn default_job_status() -> String {
40    "active".to_string()
41}
42
43fn default_max_retries() -> u32 {
44    3
45}
46
47/// SurrealDB-backed cron scheduler.
48pub struct CronScheduler {
49    memory: Option<Arc<SurrealMemory>>,
50    /// In-memory fallback for noop/testing — keyed by name
51    noop_jobs: std::sync::Mutex<Vec<CronJob>>,
52}
53
54impl CronScheduler {
55    /// Create with a SurrealDB memory backend.
56    pub fn new(memory: Arc<SurrealMemory>) -> Self {
57        Self::new_maybe(Some(memory))
58    }
59
60    /// Create without memory backend (in-memory only).
61    pub fn new_noop() -> Self {
62        Self::new_maybe(None)
63    }
64
65    fn new_maybe(memory: Option<Arc<SurrealMemory>>) -> Self {
66        Self {
67            memory,
68            noop_jobs: std::sync::Mutex::new(Vec::new()),
69        }
70    }
71
72    /// Add a new cron job. Returns the job ID.
73    pub async fn add(
74        &self,
75        name: &str,
76        schedule: &str,
77        task: &str,
78        channel: &str,
79        chat_id: &str,
80        model: &str,
81    ) -> anyhow::Result<String> {
82        // Validate cron expression
83        let parsed = cron::Schedule::from_str(schedule)
84            .map_err(|e| anyhow::anyhow!("Invalid cron expression: {}", e))?;
85
86        // A parseable expression can still have no future occurrence (a
87        // year-pinned one in the past). Storing that as an 'active' job with
88        // next_run = NONE reports success for a job the due query can never
89        // match, so refuse it at creation instead.
90        let next_run = parsed
91            .upcoming(chrono::Utc)
92            .next()
93            .map(|t| t.to_rfc3339())
94            .ok_or_else(|| {
95                anyhow::anyhow!("cron expression {schedule:?} has no future occurrences")
96            })?;
97        let next_run = Some(next_run);
98
99        let job = CronJob {
100            id: Some(name.to_string()),
101            name: name.to_string(),
102            schedule: schedule.to_string(),
103            one_shot: false,
104            task: task.to_string(),
105            channel: channel.to_string(),
106            chat_id: chat_id.to_string(),
107            model: model.to_string(),
108            enabled: true,
109            last_run: None,
110            next_run,
111            status: default_job_status(),
112            retry_count: 0,
113            max_retries: default_max_retries(),
114            last_error: None,
115            lease_until: None,
116            run_token: None,
117        };
118
119        if let Some(ref memory) = self.memory {
120            let db = memory.db().await?;
121            let created: Option<CronJob> = db.create("cron_jobs").content(job).await?;
122            let created = created.ok_or_else(|| anyhow::anyhow!("Failed to create cron job"))?;
123            Ok(created
124                .id
125                .ok_or_else(|| anyhow::anyhow!("Created cron job is missing an id"))?
126                .to_string())
127        } else {
128            let mut jobs = self.noop_jobs.lock().unwrap();
129            jobs.push(job);
130            Ok(name.to_string())
131        }
132    }
133
134    pub async fn add_once(
135        &self,
136        name: &str,
137        run_at: chrono::DateTime<chrono::Utc>,
138        task: &str,
139        channel: &str,
140        chat_id: &str,
141        model: &str,
142    ) -> anyhow::Result<String> {
143        if run_at <= chrono::Utc::now() {
144            anyhow::bail!("run_at must be in the future");
145        }
146        let job = CronJob {
147            id: Some(name.to_string()),
148            name: name.to_string(),
149            schedule: String::new(),
150            one_shot: true,
151            task: task.to_string(),
152            channel: channel.to_string(),
153            chat_id: chat_id.to_string(),
154            model: model.to_string(),
155            enabled: true,
156            last_run: None,
157            next_run: Some(run_at.to_rfc3339()),
158            status: default_job_status(),
159            retry_count: 0,
160            max_retries: default_max_retries(),
161            last_error: None,
162            lease_until: None,
163            run_token: None,
164        };
165        if let Some(ref memory) = self.memory {
166            let created: Option<CronJob> =
167                memory.db().await?.create("cron_jobs").content(job).await?;
168            Ok(created
169                .and_then(|job| job.id)
170                .ok_or_else(|| anyhow::anyhow!("Created one-shot job is missing an id"))?)
171        } else {
172            self.noop_jobs.lock().unwrap().push(job);
173            Ok(name.to_string())
174        }
175    }
176
177    /// List all cron jobs.
178    pub async fn list(&self) -> anyhow::Result<Vec<CronJob>> {
179        if let Some(ref memory) = self.memory {
180            let db = memory.db().await?;
181            let mut result: surrealdb::Response =
182                db.query("SELECT * FROM cron_jobs ORDER BY name").await?;
183            let jobs: Vec<CronJob> = result.take(0)?;
184            Ok(jobs)
185        } else {
186            let jobs = self.noop_jobs.lock().unwrap();
187            Ok(jobs.clone())
188        }
189    }
190
191    /// Remove a cron job by ID or name.
192    pub async fn remove(&self, id_or_name: &str) -> anyhow::Result<bool> {
193        if let Some(ref memory) = self.memory {
194            let db = memory.db().await?;
195            let mut result: surrealdb::Response = db
196                .query("DELETE FROM cron_jobs WHERE id = $target OR name = $target")
197                .bind(("target", id_or_name.to_string()))
198                .await?;
199            let deleted: Vec<CronJob> = result.take(0)?;
200            Ok(!deleted.is_empty())
201        } else {
202            let mut jobs = self.noop_jobs.lock().unwrap();
203            let before = jobs.len();
204            jobs.retain(|j| j.name != id_or_name);
205            Ok(jobs.len() < before)
206        }
207    }
208
209    /// Look up a job by id or name.
210    async fn find(&self, id_or_name: &str) -> anyhow::Result<Option<CronJob>> {
211        Ok(self
212            .list()
213            .await?
214            .into_iter()
215            .find(|job| job.id.as_deref() == Some(id_or_name) || job.name == id_or_name))
216    }
217
218    /// Enable a cron job.
219    ///
220    /// Enabling is the documented recovery from a dead status, so it also
221    /// recomputes `next_run` and clears the status. Setting `enabled = true`
222    /// alone leaves a job matching no branch of the due query while `list`
223    /// reports it as enabled.
224    pub async fn enable(&self, id_or_name: &str) -> anyhow::Result<bool> {
225        let Some(job) = self.find(id_or_name).await? else {
226            return Ok(false);
227        };
228
229        // A running job owns a lease; do not disturb its schedule state.
230        let revived = if job.status == "running" {
231            None
232        } else if job.one_shot || job.schedule.is_empty() {
233            let pending = job
234                .next_run
235                .as_deref()
236                .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
237                .is_some_and(|t| t > chrono::Utc::now());
238            if !pending {
239                anyhow::bail!(
240                    "cannot enable one-shot job {id_or_name:?}: its run time has already passed"
241                );
242            }
243            Some(job.next_run.clone())
244        } else {
245            let parsed = cron::Schedule::from_str(&job.schedule).map_err(|e| {
246                anyhow::anyhow!(
247                    "cannot enable {id_or_name:?}: invalid cron expression {:?}: {e}",
248                    job.schedule
249                )
250            })?;
251            let next = parsed.upcoming(chrono::Utc).next().ok_or_else(|| {
252                anyhow::anyhow!(
253                    "cannot enable {id_or_name:?}: cron expression {:?} has no future occurrences",
254                    job.schedule
255                )
256            })?;
257            Some(Some(next.to_rfc3339()))
258        };
259
260        if let Some(ref memory) = self.memory {
261            let db = memory.db().await?;
262            let query = if revived.is_some() {
263                "UPDATE cron_jobs SET enabled = true, status = 'active', next_run = $next, retry_count = 0, last_error = NONE WHERE id = $target OR name = $target"
264            } else {
265                "UPDATE cron_jobs SET enabled = true WHERE id = $target OR name = $target"
266            };
267            let mut result: surrealdb::Response = db
268                .query(query)
269                .bind(("target", id_or_name.to_string()))
270                .bind(("next", revived.clone().unwrap_or_default()))
271                .await?;
272            let updated: Vec<CronJob> = result.take(0)?;
273            Ok(!updated.is_empty())
274        } else {
275            let mut jobs = self.noop_jobs.lock().unwrap();
276            let Some(job) = jobs
277                .iter_mut()
278                .find(|j| j.id.as_deref() == Some(id_or_name) || j.name == id_or_name)
279            else {
280                return Ok(false);
281            };
282            job.enabled = true;
283            if let Some(next_run) = revived {
284                job.next_run = next_run;
285                job.status = "active".to_string();
286                job.retry_count = 0;
287                job.last_error = None;
288            }
289            Ok(true)
290        }
291    }
292
293    /// Disable a cron job.
294    pub async fn disable(&self, id_or_name: &str) -> anyhow::Result<bool> {
295        if let Some(ref memory) = self.memory {
296            let db = memory.db().await?;
297            let mut result: surrealdb::Response = db
298                .query("UPDATE cron_jobs SET enabled = false WHERE id = $target OR name = $target")
299                .bind(("target", id_or_name.to_string()))
300                .await?;
301            let updated: Vec<CronJob> = result.take(0)?;
302            Ok(!updated.is_empty())
303        } else {
304            let mut jobs = self.noop_jobs.lock().unwrap();
305            if let Some(job) = jobs
306                .iter_mut()
307                .find(|j| j.id.as_deref() == Some(id_or_name) || j.name == id_or_name)
308            {
309                job.enabled = false;
310                Ok(true)
311            } else {
312                Ok(false)
313            }
314        }
315    }
316
317    /// Get due jobs (next_run <= now, enabled).
318    pub async fn due_jobs(&self) -> anyhow::Result<Vec<CronJob>> {
319        let now = chrono::Utc::now().to_rfc3339();
320        if let Some(ref memory) = self.memory {
321            let db = memory.db().await?;
322            let mut result: surrealdb::Response = db
323                .query("SELECT * FROM cron_jobs WHERE enabled = true AND ((status = 'active' OR status = NONE) AND next_run != NONE AND next_run <= $now OR status = 'running' AND lease_until != NONE AND lease_until <= $now)")
324                .bind(("now", now))
325                .await?;
326            let jobs: Vec<CronJob> = result.take(0)?;
327            Ok(jobs)
328        } else {
329            let jobs = self.noop_jobs.lock().unwrap();
330            Ok(jobs
331                .iter()
332                .filter(|j| {
333                    j.enabled
334                        && ((j.status == "active" && j.next_run.as_deref() <= Some(&now))
335                            || (j.status == "running" && j.lease_until.as_deref() <= Some(&now)))
336                })
337                .cloned()
338                .collect())
339        }
340    }
341
342    pub async fn claim_due_jobs(&self, channel: &str) -> anyhow::Result<Vec<CronJob>> {
343        let due = self.due_jobs().await?;
344        let claim_now = chrono::Utc::now().to_rfc3339();
345        let lease_until = (chrono::Utc::now() + chrono::Duration::minutes(10)).to_rfc3339();
346        let mut claimed = Vec::new();
347        for job in due {
348            if job.channel != channel {
349                continue;
350            }
351            let Some(job_id) = job.id.as_deref() else {
352                continue;
353            };
354            let run_token = uuid::Uuid::new_v4().to_string();
355            if let Some(ref memory) = self.memory {
356                let mut result = memory
357                    .db().await?
358                    .query("UPDATE cron_jobs SET status = 'running', lease_until = $lease, run_token = $run_token WHERE id = $id AND enabled = true AND (status = 'active' OR status = NONE OR status = 'running' AND lease_until <= $now) RETURN AFTER")
359                    .bind(("id", job_id.to_string()))
360                    .bind(("lease", lease_until.clone()))
361                    .bind(("run_token", run_token.clone()))
362                    .bind(("now", claim_now.clone()))
363                    .await?;
364                let updated: Vec<CronJob> = result.take(0)?;
365                claimed.extend(updated);
366            } else {
367                let mut jobs = self.noop_jobs.lock().unwrap();
368                if let Some(stored) = jobs.iter_mut().find(|stored| {
369                    stored.id.as_deref() == Some(job_id)
370                        && (stored.status == "active"
371                            || stored.lease_until.as_deref() <= Some(claim_now.as_str()))
372                }) {
373                    stored.status = "running".to_string();
374                    stored.lease_until = Some(lease_until.clone());
375                    stored.run_token = Some(run_token);
376                    claimed.push(stored.clone());
377                }
378            }
379        }
380        Ok(claimed)
381    }
382
383    /// Mark a job as just run and compute next_run.
384    pub async fn mark_run(
385        &self,
386        job_id: &str,
387        run_token: &str,
388        schedule: &str,
389    ) -> anyhow::Result<()> {
390        let now = chrono::Utc::now();
391        // A job with no computable next run must not stay 'active': the
392        // due-jobs query requires next_run != NONE, so it would be silently
393        // unschedulable forever with nothing recorded.
394        let (next_run, status, last_error) = if schedule.is_empty() {
395            (None, "completed", None)
396        } else {
397            match cron::Schedule::from_str(schedule) {
398                Err(e) => {
399                    let reason = format!("invalid cron expression {schedule:?}: {e}");
400                    tracing::error!("cron job {}: {}", job_id, reason);
401                    (None, "invalid_schedule", Some(reason))
402                }
403                Ok(parsed) => match parsed.upcoming(chrono::Utc).next() {
404                    Some(next) => (Some(next.to_rfc3339()), "active", None),
405                    None => {
406                        let reason =
407                            format!("cron expression {schedule:?} has no future occurrences");
408                        tracing::info!("cron job {}: {}", job_id, reason);
409                        (None, "exhausted", Some(reason))
410                    }
411                },
412            }
413        };
414
415        if let Some(ref memory) = self.memory {
416            let db = memory.db().await?;
417            let mut result = db
418                .query("UPDATE cron_jobs SET last_run = $last, next_run = $next, status = $status, retry_count = 0, last_error = $last_error, lease_until = NONE, run_token = NONE WHERE id = $id AND status = 'running' AND run_token = $run_token")
419                .bind(("last", now.to_rfc3339()))
420                .bind(("next", next_run))
421                .bind(("status", status.to_string()))
422                .bind(("last_error", last_error.clone()))
423                .bind(("id", job_id.to_string()))
424                .bind(("run_token", run_token.to_string()))
425                .await?;
426            let _: Vec<CronJob> = result.take(0)?;
427        } else {
428            let mut jobs = self.noop_jobs.lock().unwrap();
429            if let Some(job) = jobs.iter_mut().find(|job| {
430                job.id.as_deref() == Some(job_id) && job.run_token.as_deref() == Some(run_token)
431            }) {
432                job.last_run = Some(now.to_rfc3339());
433                job.next_run = next_run;
434                job.status = status.to_string();
435                job.retry_count = 0;
436                job.last_error = last_error.clone();
437                job.lease_until = None;
438                job.run_token = None;
439            }
440        }
441        Ok(())
442    }
443
444    pub async fn release_run(&self, job_id: &str, run_token: &str) -> anyhow::Result<()> {
445        if let Some(ref memory) = self.memory {
446            let mut result = memory
447                .db().await?
448                .query("UPDATE cron_jobs SET status = 'active', lease_until = NONE, run_token = NONE WHERE id = $id AND status = 'running' AND run_token = $run_token")
449                .bind(("id", job_id.to_string()))
450                .bind(("run_token", run_token.to_string()))
451                .await?;
452            let _: Vec<CronJob> = result.take(0)?;
453        } else {
454            let mut jobs = self.noop_jobs.lock().unwrap();
455            if let Some(job) = jobs.iter_mut().find(|job| {
456                job.id.as_deref() == Some(job_id) && job.run_token.as_deref() == Some(run_token)
457            }) {
458                job.status = "active".to_string();
459                job.lease_until = None;
460                job.run_token = None;
461            }
462        }
463        Ok(())
464    }
465
466    pub async fn fail_run(&self, job_id: &str, run_token: &str, error: &str) -> anyhow::Result<()> {
467        let redacted = crate::redaction::redact_text(error);
468        if let Some(ref memory) = self.memory {
469            let db = memory.db().await?;
470            let Some(job) = self.list().await?.into_iter().find(|job| {
471                job.id.as_deref() == Some(job_id) && job.run_token.as_deref() == Some(run_token)
472            }) else {
473                return Ok(());
474            };
475            let retry_count = job.retry_count.saturating_add(1);
476            let status = if retry_count >= job.max_retries {
477                "failed"
478            } else {
479                "active"
480            };
481            let delay = 2_i64.saturating_pow(retry_count.min(10)) * 60;
482            let next_run = (chrono::Utc::now() + chrono::Duration::seconds(delay)).to_rfc3339();
483            let mut result = db
484                .query("UPDATE cron_jobs SET status = $status, retry_count = $retry_count, last_error = $error, next_run = $next_run, lease_until = NONE, run_token = NONE WHERE id = $id AND status = 'running' AND run_token = $run_token")
485                .bind(("status", status.to_string()))
486                .bind(("retry_count", retry_count))
487                .bind(("error", redacted))
488                .bind(("next_run", next_run))
489                .bind(("id", job_id.to_string()))
490                .bind(("run_token", run_token.to_string()))
491                .await?;
492            let _: Vec<CronJob> = result.take(0)?;
493        } else {
494            let mut jobs = self.noop_jobs.lock().unwrap();
495            if let Some(job) = jobs.iter_mut().find(|job| {
496                job.id.as_deref() == Some(job_id) && job.run_token.as_deref() == Some(run_token)
497            }) {
498                job.retry_count = job.retry_count.saturating_add(1);
499                job.status = if job.retry_count >= job.max_retries {
500                    "failed".to_string()
501                } else {
502                    "active".to_string()
503                };
504                job.last_error = Some(redacted);
505                job.lease_until = None;
506                job.run_token = None;
507            }
508        }
509        Ok(())
510    }
511}
512
513/// A due job ready to execute (returned by the ticker).
514#[derive(Debug, Clone)]
515pub struct DueJob {
516    pub job: CronJob,
517}
518
519/// Start the cron ticker as a background task. Returns a receiver for due jobs.
520pub fn start_cron_ticker(
521    scheduler: Arc<CronScheduler>,
522    channel: String,
523) -> (
524    tokio::sync::mpsc::Receiver<DueJob>,
525    Arc<tokio::sync::Notify>,
526) {
527    let (tx, rx) = tokio::sync::mpsc::channel(32);
528    let shutdown = Arc::new(tokio::sync::Notify::new());
529    let shutdown_clone = shutdown.clone();
530
531    tokio::spawn(async move {
532        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(60));
533        // Skip immediate first tick
534        interval.tick().await;
535
536        loop {
537            tokio::select! {
538                _ = interval.tick() => {
539                    match scheduler.claim_due_jobs(&channel).await {
540                        Ok(jobs) => {
541                            for job in jobs {
542                                tracing::info!("Cron: job '{}' is due", job.name);
543
544                                if tx.send(DueJob { job }).await.is_err() {
545                                    return; // Receiver dropped
546                                }
547                            }
548                        }
549                        Err(e) => {
550                            tracing::error!("Cron ticker error: {}", e);
551                        }
552                    }
553                }
554                _ = shutdown_clone.notified() => {
555                    tracing::info!("Cron ticker: shutting down");
556                    break;
557                }
558            }
559        }
560    });
561
562    (rx, shutdown)
563}
564
565#[cfg(test)]
566mod tests {
567    use super::*;
568
569    fn due_job(channel: &str) -> CronJob {
570        CronJob {
571            id: Some("job-1".to_string()),
572            name: "job".to_string(),
573            schedule: "0 0 0 * * * *".to_string(),
574            one_shot: false,
575            task: "task".to_string(),
576            channel: channel.to_string(),
577            chat_id: "chat".to_string(),
578            model: "model".to_string(),
579            enabled: true,
580            last_run: None,
581            next_run: Some((chrono::Utc::now() - chrono::Duration::minutes(1)).to_rfc3339()),
582            status: "active".to_string(),
583            retry_count: 0,
584            max_retries: 3,
585            last_error: None,
586            lease_until: None,
587            run_token: None,
588        }
589    }
590
591    #[tokio::test]
592    async fn claims_only_active_channel() {
593        let scheduler = CronScheduler::new_noop();
594        scheduler
595            .noop_jobs
596            .lock()
597            .unwrap()
598            .push(due_job("telegram"));
599        assert!(scheduler.claim_due_jobs("cli").await.unwrap().is_empty());
600        let claimed = scheduler.claim_due_jobs("telegram").await.unwrap();
601        assert_eq!(claimed.len(), 1);
602        assert!(claimed[0].run_token.is_some());
603    }
604
605    #[tokio::test]
606    async fn stale_run_token_cannot_complete_job() {
607        let scheduler = CronScheduler::new_noop();
608        scheduler.noop_jobs.lock().unwrap().push(due_job("cli"));
609        let job = scheduler.claim_due_jobs("cli").await.unwrap().remove(0);
610        scheduler
611            .mark_run("job-1", "stale-token", &job.schedule)
612            .await
613            .unwrap();
614        assert_eq!(scheduler.list().await.unwrap()[0].status, "running");
615        scheduler
616            .mark_run("job-1", job.run_token.as_deref().unwrap(), &job.schedule)
617            .await
618            .unwrap();
619        assert_eq!(scheduler.list().await.unwrap()[0].status, "active");
620    }
621
622    async fn mark_run_with_schedule(schedule: &str) -> CronJob {
623        let scheduler = CronScheduler::new_noop();
624        let mut job = due_job("cli");
625        job.schedule = schedule.to_string();
626        scheduler.noop_jobs.lock().unwrap().push(job);
627        let claimed = scheduler.claim_due_jobs("cli").await.unwrap().remove(0);
628        scheduler
629            .mark_run("job-1", claimed.run_token.as_deref().unwrap(), schedule)
630            .await
631            .unwrap();
632        scheduler.list().await.unwrap().remove(0)
633    }
634
635    #[tokio::test]
636    async fn exhausted_schedule_is_not_left_active() {
637        // Year-pinned in the past: validates, fires, then has no next run.
638        let job = mark_run_with_schedule("0 0 12 1 1 * 2020").await;
639        assert_eq!(job.status, "exhausted");
640        assert!(job.next_run.is_none());
641        assert!(job.last_error.is_some(), "the reason must be recorded");
642    }
643
644    #[tokio::test]
645    async fn unparseable_schedule_is_not_left_active() {
646        let job = mark_run_with_schedule("not a cron expression").await;
647        assert_eq!(job.status, "invalid_schedule");
648        assert!(job.next_run.is_none());
649        assert!(job.last_error.is_some());
650    }
651
652    #[tokio::test]
653    async fn add_refuses_a_schedule_with_no_future_occurrence() {
654        let scheduler = CronScheduler::new_noop();
655        // Year-pinned in the past: parses, but never runs again.
656        let error = scheduler
657            .add(
658                "yearly",
659                "0 0 12 1 1 * 2020",
660                "task",
661                "cli",
662                "chat",
663                "model",
664            )
665            .await
666            .expect_err("a job that can never run must not be reported as scheduled");
667        assert!(
668            error.to_string().contains("no future occurrences"),
669            "unexpected error: {error}"
670        );
671        assert!(scheduler.list().await.unwrap().is_empty());
672    }
673
674    #[tokio::test]
675    async fn enable_revives_a_job_left_with_a_dead_status() {
676        let scheduler = CronScheduler::new_noop();
677        let mut job = due_job("cli");
678        job.schedule = "0 0 * * * * *".to_string();
679        job.status = "exhausted".to_string();
680        job.next_run = None;
681        job.last_error = Some("dead".to_string());
682        scheduler.noop_jobs.lock().unwrap().push(job);
683
684        assert!(scheduler.disable("job-1").await.unwrap());
685        assert!(scheduler.enable("job-1").await.unwrap());
686
687        let job = scheduler.list().await.unwrap().remove(0);
688        assert!(job.enabled);
689        assert_eq!(
690            job.status, "active",
691            "enable must clear a recoverable status"
692        );
693        assert!(job.next_run.is_some(), "enable must recompute next_run");
694        assert!(job.last_error.is_none());
695    }
696
697    #[tokio::test]
698    async fn enable_fails_loudly_when_the_schedule_cannot_produce_a_next_run() {
699        let scheduler = CronScheduler::new_noop();
700        let mut job = due_job("cli");
701        job.schedule = "0 0 12 1 1 * 2020".to_string();
702        job.status = "exhausted".to_string();
703        job.next_run = None;
704        scheduler.noop_jobs.lock().unwrap().push(job);
705
706        let error = scheduler
707            .enable("job-1")
708            .await
709            .expect_err("a job that cannot be revived must not report success");
710        assert!(
711            error.to_string().contains("no future occurrences"),
712            "unexpected error: {error}"
713        );
714    }
715
716    #[tokio::test]
717    async fn enable_leaves_a_running_job_alone() {
718        let scheduler = CronScheduler::new_noop();
719        scheduler.noop_jobs.lock().unwrap().push(due_job("cli"));
720        let claimed = scheduler.claim_due_jobs("cli").await.unwrap().remove(0);
721        assert!(scheduler.enable("job-1").await.unwrap());
722        let job = scheduler.list().await.unwrap().remove(0);
723        assert_eq!(job.status, "running");
724        assert_eq!(job.run_token, claimed.run_token);
725    }
726
727    #[tokio::test]
728    async fn valid_schedule_stays_active_with_a_next_run() {
729        let job = mark_run_with_schedule("0 0 * * * * *").await;
730        assert_eq!(job.status, "active");
731        assert!(job.next_run.is_some());
732        assert!(job.last_error.is_none());
733    }
734}
apollo/cron_scheduler.rs

apollo/
cron_scheduler.rs
