chronon_runtime/
coordinator_service.rs

1//! Job and run CRUD for HTTP handlers and host integration.
2
3use std::sync::Arc;
4
5use chrono::Utc;
6use chronon_core::models::{Job, JobRevision, Run, RunStatus, ScheduleKind};
7use chronon_core::store::SchedulerStore;
8use chronon_core::{ChrononError, Result};
9use chronon_scheduler::{partition_hash_i64_for_job_id, CronExpr, job_execution_pool_id};
10use serde_json::Value;
11
12/// Object-safe coordinator operations backed by [`SchedulerStore`].
13///
14/// Used by `chronon-axum` handlers and embedded hosts that need job/run APIs without
15/// starting scheduler loops.
16pub struct CoordinatorService {
17    store: Arc<dyn SchedulerStore>,
18}
19
20impl CoordinatorService {
21    /// Wraps an existing store; does not start background tasks.
22    pub fn new(store: Arc<dyn SchedulerStore>) -> Self {
23        Self { store }
24    }
25
26    /// Underlying store for advanced host queries.
27    pub fn store(&self) -> Arc<dyn SchedulerStore> {
28        Arc::clone(&self.store)
29    }
30
31    /// Insert or update a job, computing partition hash and next cron fire time when needed.
32    ///
33    /// Appends a [`JobRevision`] when `current_revision` changes.
34    ///
35    /// # Examples
36    ///
37    /// ```
38    /// use std::sync::Arc;
39    /// use chronon_backend_mem::InMemorySchedulerStore;
40    /// use chronon_core::{Job, ScheduleKind};
41    /// use chronon_runtime::CoordinatorService;
42    ///
43    /// # #[tokio::main]
44    /// # async fn main() -> chronon_core::Result<()> {
45    /// let store = Arc::new(InMemorySchedulerStore::new());
46    /// let coordinator = CoordinatorService::new(store);
47    /// let mut job = Job::new("nightly", "noop");
48    /// job.schedule_kind = ScheduleKind::Manual;
49    /// coordinator.upsert_job(job).await?;
50    /// assert_eq!(coordinator.list_jobs().await?.len(), 1);
51    /// # Ok(())
52    /// # }
53    /// ```
54    ///
55    /// End-to-end with a real script: `cargo run -p uf-chronon --example script_macro --features mem`
56    /// (stringly job) or `script_handle_job` (typed `ScriptHandle` defaults).
57    pub async fn upsert_job(&self, mut job: Job) -> Result<()> {
58        if job.partition_hash.is_none() {
59            job.partition_hash = Some(partition_hash_i64_for_job_id(&job.job_id));
60        }
61        if job.schedule_kind == ScheduleKind::Cron {
62            if let Some(ref cron_expr) = job.cron_expr {
63                let cron = CronExpr::parse(cron_expr, job.timezone.as_deref())?;
64                job.next_run_at = cron.next_from_now();
65            }
66        }
67        job.updated_at = Utc::now();
68
69        if let Some(existing) = self.store.get_job(&job.job_id).await? {
70            if existing.current_revision != job.current_revision {
71                let revision = JobRevision::new(
72                    &job.job_id,
73                    job.current_revision,
74                    job.actor_json.clone(),
75                    serde_json::to_value(&job)?,
76                );
77                self.store.append_revision(&revision).await?;
78            }
79        } else {
80            let revision = JobRevision::new(
81                &job.job_id,
82                1,
83                job.actor_json.clone(),
84                serde_json::to_value(&job)?,
85            );
86            self.store.append_revision(&revision).await?;
87        }
88
89        self.store.upsert_job(&job).await
90    }
91
92    /// Load a job by stable `job_id`.
93    pub async fn get_job(&self, job_id: &str) -> Option<Job> {
94        self.store.get_job(job_id).await.ok().flatten()
95    }
96
97    /// Load a job by human-readable `job_name`.
98    pub async fn get_job_by_name(&self, job_name: &str) -> Option<Job> {
99        self.store
100            .get_job_by_name(job_name)
101            .await
102            .ok()
103            .flatten()
104    }
105
106    /// All jobs in the store.
107    ///
108    /// # Errors
109    ///
110    /// Returns a storage error when the underlying store fails.
111    pub async fn list_jobs(&self) -> Result<Vec<Job>> {
112        self.store.list_jobs().await
113    }
114
115    /// Disable scheduling for `job_id` without deleting the row.
116    pub async fn pause_job(&self, job_id: &str) -> Result<()> {
117        self.store.pause_job(job_id).await
118    }
119
120    /// Re-enable scheduling for `job_id`.
121    pub async fn resume_job(&self, job_id: &str) -> Result<()> {
122        self.store.resume_job(job_id).await
123    }
124
125    /// Paginated run listing with optional `job_id` and status string filters.
126    ///
127    /// Unrecognized status strings are ignored (no filter).
128    pub async fn list_runs(
129        &self,
130        job_id: Option<&str>,
131        status: Option<&str>,
132        offset: usize,
133        limit: usize,
134    ) -> Result<Vec<Run>> {
135        let status_filter = status.and_then(parse_run_status);
136        self.store
137            .list_runs_filtered(job_id, status_filter, offset, limit)
138            .await
139    }
140
141    /// Load a single run by `run_id`.
142    pub async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
143        self.store.get_run(run_id).await
144    }
145
146    /// Revision history for `job_id`, oldest first per store ordering.
147    pub async fn list_revisions(&self, job_id: &str) -> Result<Vec<JobRevision>> {
148        self.store.list_revisions(job_id).await
149    }
150
151    /// Enqueue an immediate run using the job's stored `params_json`.
152    ///
153    /// Returns the new `run_id`. Errors with [`ChrononError::JobNotFound`] when missing.
154    ///
155    /// Manual jobs (`ScheduleKind::Manual`) are never due for the tick loop; use this
156    /// (or HTTP `POST /jobs/run_now`) to trigger them. Works for any schedule kind.
157    ///
158    /// Runnable sample: `cargo run -p uf-chronon --example run_now --features mem`.
159    pub async fn run_now(&self, job_id: &str) -> Result<String> {
160        self.run_now_with_params(job_id, None).await
161    }
162
163    /// Enqueue an immediate run, optionally overriding params JSON.
164    pub async fn run_now_with_params(
165        &self,
166        job_id: &str,
167        params_override: Option<Value>,
168    ) -> Result<String> {
169        let Some(job) = self.store.get_job(job_id).await? else {
170            return Err(ChrononError::JobNotFound(job_id.to_string()));
171        };
172        let now = Utc::now();
173        let mut run = Run::for_job(&job.job_id, &job.script_name, now);
174        run.actor_json = job.actor_json.clone();
175        run.params_json = params_override.unwrap_or_else(|| job.params_json.clone());
176        run.pool_id = Some(job_execution_pool_id(&job));
177        let run_id = run.run_id.clone();
178        self.store.create_run(&run).await?;
179        Ok(run_id)
180    }
181}
182
183fn parse_run_status(s: &str) -> Option<RunStatus> {
184    match s.to_ascii_lowercase().as_str() {
185        "queued" => Some(RunStatus::Queued),
186        "claimed" => Some(RunStatus::Claimed),
187        "running" => Some(RunStatus::Running),
188        "success" => Some(RunStatus::Success),
189        "failed" => Some(RunStatus::Failed),
190        "canceled" => Some(RunStatus::Canceled),
191        "timeout" => Some(RunStatus::Timeout),
192        _ => None,
193    }
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199    use chronon_backend_mem::InMemorySchedulerStore;
200
201    #[tokio::test]
202    async fn upsert_and_run_now() {
203        let store = Arc::new(InMemorySchedulerStore::new());
204        let svc = CoordinatorService::new(store);
205        let job = Job::new("j1", "script_a");
206        svc.upsert_job(job.clone()).await.unwrap();
207        let run_id = svc.run_now(&job.job_id).await.unwrap();
208        assert!(!run_id.is_empty());
209    }
210}
chronon_runtime/coordinator_service.rs

chronon_runtime/
coordinator_service.rs
