scheduler/
execution_guard.rs

1use crate::error::ExecutionGuardErrorKind;
2use chrono::{DateTime, Utc};
3use std::convert::Infallible;
4use std::future::Future;
5use std::sync::Arc;
6use std::time::Duration;
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
9pub enum ExecutionGuardScope {
10    #[default]
11    Occurrence,
12    Resource,
13}
14
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct ExecutionSlot {
17    pub job_id: String,
18    pub resource_id: String,
19    pub scope: ExecutionGuardScope,
20    pub scheduled_at: Option<DateTime<Utc>>,
21}
22
23impl ExecutionSlot {
24    pub fn new(job_id: impl Into<String>, scheduled_at: DateTime<Utc>) -> Self {
25        let job_id = job_id.into();
26        Self::for_occurrence(job_id.clone(), job_id, scheduled_at)
27    }
28
29    pub fn for_occurrence(
30        job_id: impl Into<String>,
31        resource_id: impl Into<String>,
32        scheduled_at: DateTime<Utc>,
33    ) -> Self {
34        Self {
35            job_id: job_id.into(),
36            resource_id: resource_id.into(),
37            scope: ExecutionGuardScope::Occurrence,
38            scheduled_at: Some(scheduled_at),
39        }
40    }
41
42    pub fn for_resource(job_id: impl Into<String>, resource_id: impl Into<String>) -> Self {
43        Self {
44            job_id: job_id.into(),
45            resource_id: resource_id.into(),
46            scope: ExecutionGuardScope::Resource,
47            scheduled_at: None,
48        }
49    }
50}
51
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct ExecutionLease {
54    pub job_id: String,
55    pub resource_id: String,
56    pub scope: ExecutionGuardScope,
57    pub scheduled_at: Option<DateTime<Utc>>,
58    pub token: String,
59    pub lease_key: String,
60}
61
62impl ExecutionLease {
63    pub fn new(
64        job_id: impl Into<String>,
65        resource_id: impl Into<String>,
66        scope: ExecutionGuardScope,
67        scheduled_at: Option<DateTime<Utc>>,
68        token: impl Into<String>,
69        lease_key: impl Into<String>,
70    ) -> Self {
71        Self {
72            job_id: job_id.into(),
73            resource_id: resource_id.into(),
74            scope,
75            scheduled_at,
76            token: token.into(),
77            lease_key: lease_key.into(),
78        }
79    }
80}
81
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub enum ExecutionGuardAcquire {
84    Acquired(ExecutionLease),
85    Contended,
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub enum ExecutionGuardRenewal {
90    Renewed,
91    Lost,
92}
93
94#[derive(Debug, Clone, PartialEq, Eq)]
95pub enum ExecutionGuardEvent {
96    Degraded { error: String },
97    Recovered,
98}
99
100pub trait ExecutionGuard {
101    type Error: std::error::Error + Send + Sync + 'static;
102
103    fn acquire(
104        &self,
105        slot: ExecutionSlot,
106    ) -> impl Future<Output = Result<ExecutionGuardAcquire, Self::Error>> + Send;
107
108    fn renew(
109        &self,
110        lease: &ExecutionLease,
111    ) -> impl Future<Output = Result<ExecutionGuardRenewal, Self::Error>> + Send;
112
113    fn release(
114        &self,
115        lease: &ExecutionLease,
116    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
117
118    fn drain_events(
119        &self,
120    ) -> impl Future<Output = Result<Vec<ExecutionGuardEvent>, Self::Error>> + Send {
121        async { Ok(Vec::new()) }
122    }
123
124    fn classify_error(_error: &Self::Error) -> ExecutionGuardErrorKind
125    where
126        Self: Sized,
127    {
128        ExecutionGuardErrorKind::Unknown
129    }
130
131    fn renew_interval(&self, _lease: &ExecutionLease) -> Option<Duration> {
132        None
133    }
134}
135
136#[derive(Debug, Clone, Copy, Default)]
137pub struct NoopExecutionGuard;
138
139impl ExecutionGuard for NoopExecutionGuard {
140    type Error = Infallible;
141
142    async fn acquire(&self, slot: ExecutionSlot) -> Result<ExecutionGuardAcquire, Self::Error> {
143        Ok(ExecutionGuardAcquire::Acquired(ExecutionLease::new(
144            slot.job_id,
145            slot.resource_id,
146            slot.scope,
147            slot.scheduled_at,
148            "",
149            "",
150        )))
151    }
152
153    async fn renew(&self, _lease: &ExecutionLease) -> Result<ExecutionGuardRenewal, Self::Error> {
154        Ok(ExecutionGuardRenewal::Renewed)
155    }
156
157    async fn release(&self, _lease: &ExecutionLease) -> Result<(), Self::Error> {
158        Ok(())
159    }
160}
161
162impl<T> ExecutionGuard for Arc<T>
163where
164    T: ExecutionGuard + Send + Sync,
165{
166    type Error = T::Error;
167
168    async fn acquire(&self, slot: ExecutionSlot) -> Result<ExecutionGuardAcquire, Self::Error> {
169        self.as_ref().acquire(slot).await
170    }
171
172    async fn renew(&self, lease: &ExecutionLease) -> Result<ExecutionGuardRenewal, Self::Error> {
173        self.as_ref().renew(lease).await
174    }
175
176    async fn release(&self, lease: &ExecutionLease) -> Result<(), Self::Error> {
177        self.as_ref().release(lease).await
178    }
179
180    async fn drain_events(&self) -> Result<Vec<ExecutionGuardEvent>, Self::Error> {
181        self.as_ref().drain_events().await
182    }
183
184    fn classify_error(error: &Self::Error) -> ExecutionGuardErrorKind
185    where
186        Self: Sized,
187    {
188        T::classify_error(error)
189    }
190
191    fn renew_interval(&self, lease: &ExecutionLease) -> Option<Duration> {
192        self.as_ref().renew_interval(lease)
193    }
194}
scheduler/execution_guard.rs

scheduler/
execution_guard.rs
