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 reclaim_expired_run_leases(&self, now: DateTime<Utc>) -> Result<Vec<String>> {
116        runs::reclaim_expired_run_leases(self, now).await
117    }
118
119    async fn append_revision(&self, revision: &JobRevision) -> Result<()> {
120        coordinator::append_revision(self, revision).await
121    }
122
123    async fn list_revisions(&self, job_id: &str) -> Result<Vec<JobRevision>> {
124        coordinator::list_revisions(self, job_id).await
125    }
126
127    async fn upsert_script(&self, script: &Script) -> Result<()> {
128        coordinator::upsert_script(self, script).await
129    }
130
131    async fn get_script(&self, script_name: &str) -> Result<Option<Script>> {
132        coordinator::get_script(self, script_name).await
133    }
134
135    async fn try_claim_run_once(
136        &self,
137        job_id: &str,
138        claimed_by: &str,
139        now: DateTime<Utc>,
140        claim_ttl_secs: i64,
141    ) -> Result<bool> {
142        claims::try_claim_run_once(self, job_id, claimed_by, now, claim_ttl_secs).await
143    }
144
145    async fn mark_run_once_completed(
146        &self,
147        job_id: &str,
148        completed_at: DateTime<Utc>,
149    ) -> Result<()> {
150        claims::mark_run_once_completed(self, job_id, completed_at).await
151    }
152
153    async fn release_run_once_claim(
154        &self,
155        job_id: &str,
156        claimed_by: &str,
157        now: DateTime<Utc>,
158    ) -> Result<()> {
159        claims::release_run_once_claim(self, job_id, claimed_by, now).await
160    }
161
162    async fn find_due_job_ids_in_partitions(
163        &self,
164        owned_partitions: &[u32],
165        due_until: DateTime<Utc>,
166        limit: u32,
167    ) -> Result<Vec<String>> {
168        claims::find_due_job_ids_in_partitions(self, owned_partitions, due_until, limit).await
169    }
170
171    async fn min_next_run_at_in_partitions(
172        &self,
173        owned_partitions: &[u32],
174    ) -> Result<Option<DateTime<Utc>>> {
175        claims::min_next_run_at_in_partitions(self, owned_partitions).await
176    }
177
178    async fn claim_job_for_tick(
179        &self,
180        job_id: &str,
181        claim_id: &str,
182        now: DateTime<Utc>,
183        lease_ttl_secs: i64,
184    ) -> Result<bool> {
185        claims::claim_job_for_tick(self, job_id, claim_id, now, lease_ttl_secs).await
186    }
187
188    async fn release_job_tick_claim(&self, job_id: &str) -> Result<()> {
189        claims::release_job_tick_claim(self, job_id).await
190    }
191
192    async fn persist_post_tick_job_state(
193        &self,
194        job_id: &str,
195        next_run_at: Option<DateTime<Utc>>,
196    ) -> Result<()> {
197        claims::persist_post_tick_job_state(self, job_id, next_run_at).await
198    }
199
200    async fn try_acquire_leader(&self, instance_id: &str, ttl_secs: i64) -> Result<bool> {
201        coordinator::try_acquire_leader(self, instance_id, ttl_secs).await
202    }
203
204    async fn renew_leader_lease(&self, instance_id: &str, ttl_secs: i64) -> Result<()> {
205        coordinator::renew_leader_lease(self, instance_id, ttl_secs).await
206    }
207
208    async fn get_leader(&self) -> Result<Option<SchedulerLeader>> {
209        coordinator::get_leader(self).await
210    }
211
212    async fn upsert_partition_assignment(&self, assignment: &PartitionAssignment) -> Result<()> {
213        coordinator::upsert_partition_assignment(self, assignment).await
214    }
215
216    async fn list_partition_assignments(&self) -> Result<Vec<PartitionAssignment>> {
217        coordinator::list_partition_assignments(self).await
218    }
219
220    async fn register_worker(&self, worker: &Worker) -> Result<()> {
221        coordinator::register_worker(self, worker).await
222    }
223
224    async fn heartbeat_worker(&self, worker_id: &str, at: DateTime<Utc>) -> Result<()> {
225        coordinator::heartbeat_worker(self, worker_id, at).await
226    }
227}
chronon_backend_sql_common/store_impl.rs

chronon_backend_sql_common/
store_impl.rs
