chronon_backend_sql_common/
store_impl.rs

1//! [`SchedulerStore`](chronon_core::store::SchedulerStore) trait surface for [`SqlSchedulerStore`].
2//!
3//! Internal — delegates to [`jobs`](crate::jobs), [`runs`](crate::runs), [`claims`](crate::claims),
4//! and [`coordinator`](crate::coordinator) modules.
5
6use async_trait::async_trait;
7use chrono::{DateTime, Utc};
8
9use chronon_core::models::{
10    Job, JobRevision, PartitionAssignment, Run, RunStatus, SchedulerLeader, Script, Worker,
11};
12use chronon_core::store::SchedulerStore;
13use chronon_core::Result;
14
15use crate::{claims, coordinator, jobs, runs, SqlSchedulerStore};
16
17#[async_trait]
18impl SchedulerStore for SqlSchedulerStore {
19    async fn upsert_job(&self, job: &Job) -> Result<()> {
20        jobs::upsert_job(self, job).await
21    }
22
23    async fn get_job(&self, job_id: &str) -> Result<Option<Job>> {
24        jobs::get_job(self, job_id).await
25    }
26
27    async fn get_job_by_name(&self, job_name: &str) -> Result<Option<Job>> {
28        jobs::get_job_by_name(self, job_name).await
29    }
30
31    async fn list_jobs(&self) -> Result<Vec<Job>> {
32        jobs::list_jobs(self).await
33    }
34
35    async fn list_due_jobs(&self, before: DateTime<Utc>) -> Result<Vec<Job>> {
36        jobs::list_due_jobs(self, before).await
37    }
38
39    async fn pause_job(&self, job_id: &str) -> Result<()> {
40        jobs::pause_job(self, job_id).await
41    }
42
43    async fn resume_job(&self, job_id: &str) -> Result<()> {
44        jobs::resume_job(self, job_id).await
45    }
46
47    async fn create_run(&self, run: &Run) -> Result<()> {
48        runs::create_run(self, run).await
49    }
50
51    async fn update_run(&self, run: &Run) -> Result<()> {
52        runs::update_run(self, run).await
53    }
54
55    async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
56        runs::get_run(self, run_id).await
57    }
58
59    async fn list_runs_for_job(&self, job_id: &str, limit: usize) -> Result<Vec<Run>> {
60        runs::list_runs_for_job(self, job_id, limit).await
61    }
62
63    async fn list_runs_filtered(
64        &self,
65        job_id: Option<&str>,
66        status: Option<RunStatus>,
67        offset: usize,
68        limit: usize,
69    ) -> Result<Vec<Run>> {
70        runs::list_runs_filtered(self, job_id, status, offset, limit).await
71    }
72
73    async fn claim_next_queued(
74        &self,
75        pool_id: &str,
76        worker_id: &str,
77        now: DateTime<Utc>,
78        lease_ttl_secs: i64,
79    ) -> Result<Option<Run>> {
80        runs::claim_next_queued(self, pool_id, worker_id, now, lease_ttl_secs).await
81    }
82
83    async fn claim_run_by_id(
84        &self,
85        run_id: &str,
86        pool_id: &str,
87        worker_id: &str,
88        now: DateTime<Utc>,
89        lease_ttl_secs: i64,
90    ) -> Result<Option<Run>> {
91        runs::claim_run_by_id(self, run_id, pool_id, worker_id, now, lease_ttl_secs).await
92    }
93
94    async fn claim_runs_by_ids(
95        &self,
96        run_ids: &[&str],
97        pool_id: &str,
98        worker_id: &str,
99        now: DateTime<Utc>,
100        lease_ttl_secs: i64,
101    ) -> Result<Vec<Run>> {
102        runs::claim_runs_by_ids(self, run_ids, pool_id, worker_id, now, lease_ttl_secs).await
103    }
104
105    async fn renew_run_lease(
106        &self,
107        run_id: &str,
108        worker_id: &str,
109        now: DateTime<Utc>,
110        lease_ttl_secs: i64,
111    ) -> Result<bool> {
112        runs::renew_run_lease(self, run_id, worker_id, now, lease_ttl_secs).await
113    }
114
115    async fn append_revision(&self, revision: &JobRevision) -> Result<()> {
116        coordinator::append_revision(self, revision).await
117    }
118
119    async fn list_revisions(&self, job_id: &str) -> Result<Vec<JobRevision>> {
120        coordinator::list_revisions(self, job_id).await
121    }
122
123    async fn upsert_script(&self, script: &Script) -> Result<()> {
124        coordinator::upsert_script(self, script).await
125    }
126
127    async fn get_script(&self, script_name: &str) -> Result<Option<Script>> {
128        coordinator::get_script(self, script_name).await
129    }
130
131    async fn try_claim_run_once(
132        &self,
133        job_id: &str,
134        claimed_by: &str,
135        now: DateTime<Utc>,
136        claim_ttl_secs: i64,
137    ) -> Result<bool> {
138        claims::try_claim_run_once(self, job_id, claimed_by, now, claim_ttl_secs).await
139    }
140
141    async fn mark_run_once_completed(
142        &self,
143        job_id: &str,
144        completed_at: DateTime<Utc>,
145    ) -> Result<()> {
146        claims::mark_run_once_completed(self, job_id, completed_at).await
147    }
148
149    async fn release_run_once_claim(
150        &self,
151        job_id: &str,
152        claimed_by: &str,
153        now: DateTime<Utc>,
154    ) -> Result<()> {
155        claims::release_run_once_claim(self, job_id, claimed_by, now).await
156    }
157
158    async fn find_due_job_ids_in_partitions(
159        &self,
160        owned_partitions: &[u32],
161        due_until: DateTime<Utc>,
162        limit: u32,
163    ) -> Result<Vec<String>> {
164        claims::find_due_job_ids_in_partitions(self, owned_partitions, due_until, limit).await
165    }
166
167    async fn min_next_run_at_in_partitions(
168        &self,
169        owned_partitions: &[u32],
170    ) -> Result<Option<DateTime<Utc>>> {
171        claims::min_next_run_at_in_partitions(self, owned_partitions).await
172    }
173
174    async fn claim_job_for_tick(
175        &self,
176        job_id: &str,
177        claim_id: &str,
178        now: DateTime<Utc>,
179        lease_ttl_secs: i64,
180    ) -> Result<bool> {
181        claims::claim_job_for_tick(self, job_id, claim_id, now, lease_ttl_secs).await
182    }
183
184    async fn release_job_tick_claim(&self, job_id: &str) -> Result<()> {
185        claims::release_job_tick_claim(self, job_id).await
186    }
187
188    async fn persist_post_tick_job_state(
189        &self,
190        job_id: &str,
191        next_run_at: Option<DateTime<Utc>>,
192    ) -> Result<()> {
193        claims::persist_post_tick_job_state(self, job_id, next_run_at).await
194    }
195
196    async fn try_acquire_leader(&self, instance_id: &str, ttl_secs: i64) -> Result<bool> {
197        coordinator::try_acquire_leader(self, instance_id, ttl_secs).await
198    }
199
200    async fn renew_leader_lease(&self, instance_id: &str, ttl_secs: i64) -> Result<()> {
201        coordinator::renew_leader_lease(self, instance_id, ttl_secs).await
202    }
203
204    async fn get_leader(&self) -> Result<Option<SchedulerLeader>> {
205        coordinator::get_leader(self).await
206    }
207
208    async fn upsert_partition_assignment(&self, assignment: &PartitionAssignment) -> Result<()> {
209        coordinator::upsert_partition_assignment(self, assignment).await
210    }
211
212    async fn list_partition_assignments(&self) -> Result<Vec<PartitionAssignment>> {
213        coordinator::list_partition_assignments(self).await
214    }
215
216    async fn register_worker(&self, worker: &Worker) -> Result<()> {
217        coordinator::register_worker(self, worker).await
218    }
219
220    async fn heartbeat_worker(&self, worker_id: &str, at: DateTime<Utc>) -> Result<()> {
221        coordinator::heartbeat_worker(self, worker_id, at).await
222    }
223}
chronon_backend_sql_common/store_impl.rs

chronon_backend_sql_common/
store_impl.rs
