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