chronon_backend_sql_common/
delegate.rs1#[macro_export]
5macro_rules! delegate_scheduler_store {
6 ($wrapper:ty, $field:ident) => {
7 #[::async_trait::async_trait]
8 impl ::chronon_core::store::SchedulerStore for $wrapper {
9 async fn upsert_job(
10 &self,
11 job: &::chronon_core::models::Job,
12 ) -> ::chronon_core::Result<()> {
13 self.$field.upsert_job(job).await
14 }
15
16 async fn get_job(
17 &self,
18 job_id: &str,
19 ) -> ::chronon_core::Result<Option<::chronon_core::models::Job>> {
20 self.$field.get_job(job_id).await
21 }
22
23 async fn get_job_by_name(
24 &self,
25 job_name: &str,
26 ) -> ::chronon_core::Result<Option<::chronon_core::models::Job>> {
27 self.$field.get_job_by_name(job_name).await
28 }
29
30 async fn list_jobs(&self) -> ::chronon_core::Result<Vec<::chronon_core::models::Job>> {
31 self.$field.list_jobs().await
32 }
33
34 async fn list_due_jobs(
35 &self,
36 before: ::chrono::DateTime<::chrono::Utc>,
37 ) -> ::chronon_core::Result<Vec<::chronon_core::models::Job>> {
38 self.$field.list_due_jobs(before).await
39 }
40
41 async fn pause_job(&self, job_id: &str) -> ::chronon_core::Result<()> {
42 self.$field.pause_job(job_id).await
43 }
44
45 async fn resume_job(&self, job_id: &str) -> ::chronon_core::Result<()> {
46 self.$field.resume_job(job_id).await
47 }
48
49 async fn create_run(
50 &self,
51 run: &::chronon_core::models::Run,
52 ) -> ::chronon_core::Result<()> {
53 self.$field.create_run(run).await
54 }
55
56 async fn update_run(
57 &self,
58 run: &::chronon_core::models::Run,
59 ) -> ::chronon_core::Result<()> {
60 self.$field.update_run(run).await
61 }
62
63 async fn get_run(
64 &self,
65 run_id: &str,
66 ) -> ::chronon_core::Result<Option<::chronon_core::models::Run>> {
67 self.$field.get_run(run_id).await
68 }
69
70 async fn list_runs_for_job(
71 &self,
72 job_id: &str,
73 limit: usize,
74 ) -> ::chronon_core::Result<Vec<::chronon_core::models::Run>> {
75 self.$field.list_runs_for_job(job_id, limit).await
76 }
77
78 async fn list_runs_filtered(
79 &self,
80 job_id: Option<&str>,
81 status: Option<::chronon_core::models::RunStatus>,
82 offset: usize,
83 limit: usize,
84 ) -> ::chronon_core::Result<Vec<::chronon_core::models::Run>> {
85 self.$field
86 .list_runs_filtered(job_id, status, offset, limit)
87 .await
88 }
89
90 async fn claim_next_queued(
91 &self,
92 pool_id: &str,
93 worker_id: &str,
94 now: ::chrono::DateTime<::chrono::Utc>,
95 lease_ttl_secs: i64,
96 ) -> ::chronon_core::Result<Option<::chronon_core::models::Run>> {
97 self.$field
98 .claim_next_queued(pool_id, worker_id, now, lease_ttl_secs)
99 .await
100 }
101
102 async fn claim_run_by_id(
103 &self,
104 run_id: &str,
105 pool_id: &str,
106 worker_id: &str,
107 now: ::chrono::DateTime<::chrono::Utc>,
108 lease_ttl_secs: i64,
109 ) -> ::chronon_core::Result<Option<::chronon_core::models::Run>> {
110 self.$field
111 .claim_run_by_id(run_id, pool_id, worker_id, now, lease_ttl_secs)
112 .await
113 }
114
115 async fn claim_runs_by_ids(
116 &self,
117 run_ids: &[&str],
118 pool_id: &str,
119 worker_id: &str,
120 now: ::chrono::DateTime<::chrono::Utc>,
121 lease_ttl_secs: i64,
122 ) -> ::chronon_core::Result<Vec<::chronon_core::models::Run>> {
123 self.$field
124 .claim_runs_by_ids(run_ids, pool_id, worker_id, now, lease_ttl_secs)
125 .await
126 }
127
128 async fn renew_run_lease(
129 &self,
130 run_id: &str,
131 worker_id: &str,
132 now: ::chrono::DateTime<::chrono::Utc>,
133 lease_ttl_secs: i64,
134 ) -> ::chronon_core::Result<bool> {
135 self.$field
136 .renew_run_lease(run_id, worker_id, now, lease_ttl_secs)
137 .await
138 }
139
140 async fn reclaim_expired_run_leases(
141 &self,
142 now: ::chrono::DateTime<::chrono::Utc>,
143 ) -> ::chronon_core::Result<Vec<String>> {
144 self.$field.reclaim_expired_run_leases(now).await
145 }
146
147 async fn append_revision(
148 &self,
149 revision: &::chronon_core::models::JobRevision,
150 ) -> ::chronon_core::Result<()> {
151 self.$field.append_revision(revision).await
152 }
153
154 async fn list_revisions(
155 &self,
156 job_id: &str,
157 ) -> ::chronon_core::Result<Vec<::chronon_core::models::JobRevision>> {
158 self.$field.list_revisions(job_id).await
159 }
160
161 async fn upsert_script(
162 &self,
163 script: &::chronon_core::models::Script,
164 ) -> ::chronon_core::Result<()> {
165 self.$field.upsert_script(script).await
166 }
167
168 async fn get_script(
169 &self,
170 script_name: &str,
171 ) -> ::chronon_core::Result<Option<::chronon_core::models::Script>> {
172 self.$field.get_script(script_name).await
173 }
174
175 async fn try_claim_run_once(
176 &self,
177 job_id: &str,
178 claimed_by: &str,
179 now: ::chrono::DateTime<::chrono::Utc>,
180 claim_ttl_secs: i64,
181 ) -> ::chronon_core::Result<bool> {
182 self.$field
183 .try_claim_run_once(job_id, claimed_by, now, claim_ttl_secs)
184 .await
185 }
186
187 async fn mark_run_once_completed(
188 &self,
189 job_id: &str,
190 completed_at: ::chrono::DateTime<::chrono::Utc>,
191 ) -> ::chronon_core::Result<()> {
192 self.$field
193 .mark_run_once_completed(job_id, completed_at)
194 .await
195 }
196
197 async fn release_run_once_claim(
198 &self,
199 job_id: &str,
200 claimed_by: &str,
201 now: ::chrono::DateTime<::chrono::Utc>,
202 ) -> ::chronon_core::Result<()> {
203 self.$field
204 .release_run_once_claim(job_id, claimed_by, now)
205 .await
206 }
207
208 async fn find_due_job_ids_in_partitions(
209 &self,
210 owned_partitions: &[u32],
211 due_until: ::chrono::DateTime<::chrono::Utc>,
212 limit: u32,
213 ) -> ::chronon_core::Result<Vec<String>> {
214 self.$field
215 .find_due_job_ids_in_partitions(owned_partitions, due_until, limit)
216 .await
217 }
218
219 async fn min_next_run_at_in_partitions(
220 &self,
221 owned_partitions: &[u32],
222 ) -> ::chronon_core::Result<Option<::chrono::DateTime<::chrono::Utc>>> {
223 self.$field
224 .min_next_run_at_in_partitions(owned_partitions)
225 .await
226 }
227
228 async fn claim_job_for_tick(
229 &self,
230 job_id: &str,
231 claim_id: &str,
232 now: ::chrono::DateTime<::chrono::Utc>,
233 lease_ttl_secs: i64,
234 ) -> ::chronon_core::Result<bool> {
235 self.$field
236 .claim_job_for_tick(job_id, claim_id, now, lease_ttl_secs)
237 .await
238 }
239
240 async fn release_job_tick_claim(&self, job_id: &str) -> ::chronon_core::Result<()> {
241 self.$field.release_job_tick_claim(job_id).await
242 }
243
244 async fn persist_post_tick_job_state(
245 &self,
246 job_id: &str,
247 next_run_at: Option<::chrono::DateTime<::chrono::Utc>>,
248 ) -> ::chronon_core::Result<()> {
249 self.$field
250 .persist_post_tick_job_state(job_id, next_run_at)
251 .await
252 }
253
254 async fn try_acquire_leader(
255 &self,
256 instance_id: &str,
257 ttl_secs: i64,
258 ) -> ::chronon_core::Result<bool> {
259 self.$field.try_acquire_leader(instance_id, ttl_secs).await
260 }
261
262 async fn renew_leader_lease(
263 &self,
264 instance_id: &str,
265 ttl_secs: i64,
266 ) -> ::chronon_core::Result<()> {
267 self.$field.renew_leader_lease(instance_id, ttl_secs).await
268 }
269
270 async fn get_leader(
271 &self,
272 ) -> ::chronon_core::Result<Option<::chronon_core::models::SchedulerLeader>> {
273 self.$field.get_leader().await
274 }
275
276 async fn upsert_partition_assignment(
277 &self,
278 assignment: &::chronon_core::models::PartitionAssignment,
279 ) -> ::chronon_core::Result<()> {
280 self.$field.upsert_partition_assignment(assignment).await
281 }
282
283 async fn list_partition_assignments(
284 &self,
285 ) -> ::chronon_core::Result<Vec<::chronon_core::models::PartitionAssignment>> {
286 self.$field.list_partition_assignments().await
287 }
288
289 async fn register_worker(
290 &self,
291 worker: &::chronon_core::models::Worker,
292 ) -> ::chronon_core::Result<()> {
293 self.$field.register_worker(worker).await
294 }
295
296 async fn heartbeat_worker(
297 &self,
298 worker_id: &str,
299 at: ::chrono::DateTime<::chrono::Utc>,
300 ) -> ::chronon_core::Result<()> {
301 self.$field.heartbeat_worker(worker_id, at).await
302 }
303 }
304 };
305}