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