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