1use crate::error::{ExecutionGuardErrorKind, StoreErrorKind};
2use crate::execution_guard::{ExecutionGuardRenewal, ExecutionGuardScope, ExecutionLease};
3use crate::model::{JobState, TriggerSource};
4use chrono::{DateTime, Utc};
5use std::convert::Infallible;
6use std::future::Future;
7use std::sync::Arc;
8use std::time::Duration;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub struct CoordinatedLeaseConfig {
12 pub ttl: Duration,
13 pub renew_interval: Duration,
14}
15
16impl CoordinatedLeaseConfig {
17 pub fn validate(self) -> Result<Self, &'static str> {
18 if self.ttl.is_zero() {
19 return Err("coordinated lease ttl must be greater than zero");
20 }
21 if self.renew_interval.is_zero() {
22 return Err("coordinated lease renew_interval must be greater than zero");
23 }
24 if self.renew_interval >= self.ttl {
25 return Err("coordinated lease renew_interval must be less than ttl");
26 }
27 Ok(self)
28 }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct CoordinatedRuntimeState {
33 pub state: JobState,
34 pub revision: u64,
35 pub paused: bool,
36}
37
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct CoordinatedPendingTrigger {
40 pub scheduled_at: DateTime<Utc>,
41 pub catch_up: bool,
42 pub trigger_count: u32,
43 pub source: TriggerSource,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct CoordinatedClaim {
48 pub state: CoordinatedRuntimeState,
49 pub trigger: CoordinatedPendingTrigger,
50 pub lease: ExecutionLease,
51 pub replayed: bool,
52}
53
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct CoordinatedCompletion {
56 pub last_run_at: DateTime<Utc>,
57 pub last_success_at: Option<DateTime<Utc>>,
58 pub last_error: Option<String>,
59}
60
61pub trait CoordinatedStateStore {
62 type Error: std::error::Error + Send + Sync + 'static;
63
64 fn load_or_initialize(
65 &self,
66 job_id: &str,
67 initial_state: JobState,
68 ) -> impl Future<Output = Result<CoordinatedRuntimeState, Self::Error>> + Send;
69
70 fn save_state(
71 &self,
72 job_id: &str,
73 revision: u64,
74 state: &JobState,
75 ) -> impl Future<Output = Result<bool, Self::Error>> + Send;
76
77 fn reclaim_inflight(
78 &self,
79 job_id: &str,
80 resource_id: &str,
81 lease_config: CoordinatedLeaseConfig,
82 ) -> impl Future<Output = Result<Option<CoordinatedClaim>, Self::Error>> + Send;
83
84 fn claim_trigger(
85 &self,
86 job_id: &str,
87 resource_id: &str,
88 revision: u64,
89 trigger: CoordinatedPendingTrigger,
90 next_state: &JobState,
91 lease_config: CoordinatedLeaseConfig,
92 scope: ExecutionGuardScope,
93 ) -> impl Future<Output = Result<Option<CoordinatedClaim>, Self::Error>> + Send;
94
95 fn renew(
96 &self,
97 lease: &ExecutionLease,
98 lease_config: CoordinatedLeaseConfig,
99 ) -> impl Future<Output = Result<ExecutionGuardRenewal, Self::Error>> + Send;
100
101 fn complete(
102 &self,
103 job_id: &str,
104 lease: &ExecutionLease,
105 completion: CoordinatedCompletion,
106 ) -> impl Future<Output = Result<bool, Self::Error>> + Send;
107
108 fn delete(&self, job_id: &str) -> impl Future<Output = Result<(), Self::Error>> + Send;
109
110 fn pause(&self, job_id: &str) -> impl Future<Output = Result<bool, Self::Error>> + Send;
111
112 fn resume(&self, job_id: &str) -> impl Future<Output = Result<bool, Self::Error>> + Send;
113
114 fn classify_store_error(_error: &Self::Error) -> StoreErrorKind
115 where
116 Self: Sized,
117 {
118 StoreErrorKind::Unknown
119 }
120
121 fn classify_guard_error(_error: &Self::Error) -> ExecutionGuardErrorKind
122 where
123 Self: Sized,
124 {
125 ExecutionGuardErrorKind::Unknown
126 }
127}
128
129#[derive(Debug, Clone, Copy, Default)]
130pub struct NoopCoordinatedStateStore;
131
132impl CoordinatedStateStore for NoopCoordinatedStateStore {
133 type Error = Infallible;
134
135 async fn load_or_initialize(
136 &self,
137 _job_id: &str,
138 initial_state: JobState,
139 ) -> Result<CoordinatedRuntimeState, Self::Error> {
140 Ok(CoordinatedRuntimeState {
141 state: initial_state,
142 revision: 0,
143 paused: false,
144 })
145 }
146
147 async fn save_state(
148 &self,
149 _job_id: &str,
150 _revision: u64,
151 _state: &JobState,
152 ) -> Result<bool, Self::Error> {
153 Ok(true)
154 }
155
156 async fn reclaim_inflight(
157 &self,
158 _job_id: &str,
159 _resource_id: &str,
160 _lease_config: CoordinatedLeaseConfig,
161 ) -> Result<Option<CoordinatedClaim>, Self::Error> {
162 Ok(None)
163 }
164
165 async fn claim_trigger(
166 &self,
167 _job_id: &str,
168 _resource_id: &str,
169 _revision: u64,
170 _trigger: CoordinatedPendingTrigger,
171 _next_state: &JobState,
172 _lease_config: CoordinatedLeaseConfig,
173 _scope: ExecutionGuardScope,
174 ) -> Result<Option<CoordinatedClaim>, Self::Error> {
175 Ok(None)
176 }
177
178 async fn renew(
179 &self,
180 _lease: &ExecutionLease,
181 _lease_config: CoordinatedLeaseConfig,
182 ) -> Result<ExecutionGuardRenewal, Self::Error> {
183 Ok(ExecutionGuardRenewal::Lost)
184 }
185
186 async fn complete(
187 &self,
188 _job_id: &str,
189 _lease: &ExecutionLease,
190 _completion: CoordinatedCompletion,
191 ) -> Result<bool, Self::Error> {
192 Ok(true)
193 }
194
195 async fn delete(&self, _job_id: &str) -> Result<(), Self::Error> {
196 Ok(())
197 }
198
199 async fn pause(&self, _job_id: &str) -> Result<bool, Self::Error> {
200 Ok(false)
201 }
202
203 async fn resume(&self, _job_id: &str) -> Result<bool, Self::Error> {
204 Ok(false)
205 }
206}
207
208impl<T> CoordinatedStateStore for Arc<T>
209where
210 T: CoordinatedStateStore + Send + Sync,
211{
212 type Error = T::Error;
213
214 async fn load_or_initialize(
215 &self,
216 job_id: &str,
217 initial_state: JobState,
218 ) -> Result<CoordinatedRuntimeState, Self::Error> {
219 self.as_ref()
220 .load_or_initialize(job_id, initial_state)
221 .await
222 }
223
224 async fn save_state(
225 &self,
226 job_id: &str,
227 revision: u64,
228 state: &JobState,
229 ) -> Result<bool, Self::Error> {
230 self.as_ref().save_state(job_id, revision, state).await
231 }
232
233 async fn reclaim_inflight(
234 &self,
235 job_id: &str,
236 resource_id: &str,
237 lease_config: CoordinatedLeaseConfig,
238 ) -> Result<Option<CoordinatedClaim>, Self::Error> {
239 self.as_ref()
240 .reclaim_inflight(job_id, resource_id, lease_config)
241 .await
242 }
243
244 async fn claim_trigger(
245 &self,
246 job_id: &str,
247 resource_id: &str,
248 revision: u64,
249 trigger: CoordinatedPendingTrigger,
250 next_state: &JobState,
251 lease_config: CoordinatedLeaseConfig,
252 scope: ExecutionGuardScope,
253 ) -> Result<Option<CoordinatedClaim>, Self::Error> {
254 self.as_ref()
255 .claim_trigger(
256 job_id,
257 resource_id,
258 revision,
259 trigger,
260 next_state,
261 lease_config,
262 scope,
263 )
264 .await
265 }
266
267 async fn renew(
268 &self,
269 lease: &ExecutionLease,
270 lease_config: CoordinatedLeaseConfig,
271 ) -> Result<ExecutionGuardRenewal, Self::Error> {
272 self.as_ref().renew(lease, lease_config).await
273 }
274
275 async fn complete(
276 &self,
277 job_id: &str,
278 lease: &ExecutionLease,
279 completion: CoordinatedCompletion,
280 ) -> Result<bool, Self::Error> {
281 self.as_ref().complete(job_id, lease, completion).await
282 }
283
284 async fn delete(&self, job_id: &str) -> Result<(), Self::Error> {
285 self.as_ref().delete(job_id).await
286 }
287
288 async fn pause(&self, job_id: &str) -> Result<bool, Self::Error> {
289 self.as_ref().pause(job_id).await
290 }
291
292 async fn resume(&self, job_id: &str) -> Result<bool, Self::Error> {
293 self.as_ref().resume(job_id).await
294 }
295
296 fn classify_store_error(error: &Self::Error) -> StoreErrorKind
297 where
298 Self: Sized,
299 {
300 T::classify_store_error(error)
301 }
302
303 fn classify_guard_error(error: &Self::Error) -> ExecutionGuardErrorKind
304 where
305 Self: Sized,
306 {
307 T::classify_guard_error(error)
308 }
309}