scheduler/store/
mod.rs

1mod in_memory;
2mod resilient;
3
4use crate::error::StoreErrorKind;
5use crate::model::JobState;
6use std::convert::Infallible;
7use std::error::Error;
8use std::future::{Future, ready};
9use std::sync::Arc;
10
11pub use in_memory::InMemoryStateStore;
12pub use resilient::ResilientStateStore;
13
14pub trait StateStore {
15    type Error: std::error::Error + Send + Sync + 'static;
16
17    fn load(
18        &self,
19        job_id: &str,
20    ) -> impl Future<Output = Result<Option<JobState>, Self::Error>> + Send;
21
22    fn save(&self, state: &JobState) -> impl Future<Output = Result<(), Self::Error>> + Send;
23
24    fn delete(&self, _job_id: &str) -> impl Future<Output = Result<(), Self::Error>> + Send {
25        ready(Ok(()))
26    }
27
28    fn drain_events(&self) -> impl Future<Output = Result<Vec<StoreEvent>, Self::Error>> + Send {
29        ready(Ok(Vec::new()))
30    }
31
32    fn classify_error(_error: &Self::Error) -> StoreErrorKind
33    where
34        Self: Sized,
35    {
36        StoreErrorKind::Unknown
37    }
38}
39
40/// Classifies store errors that should trigger a one-way downgrade to the
41/// in-process mirror store.
42pub trait ResilientStoreError: Error + Send + Sync + 'static {
43    fn is_connection_issue(&self) -> bool;
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47pub enum StoreOperation {
48    Load,
49    Save,
50    Delete,
51}
52
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub enum StoreEvent {
55    Degraded {
56        operation: StoreOperation,
57        error: String,
58    },
59    Recovering {
60        operation: StoreOperation,
61    },
62    Recovered {
63        operation: StoreOperation,
64    },
65    RecoveryFailed {
66        operation: StoreOperation,
67        error: String,
68    },
69    RecoveryConflict {
70        operation: StoreOperation,
71        job_id: String,
72        error: String,
73    },
74}
75
76impl<T> StateStore for Arc<T>
77where
78    T: StateStore + Send + Sync,
79{
80    type Error = T::Error;
81
82    async fn load(&self, job_id: &str) -> Result<Option<JobState>, Self::Error> {
83        self.as_ref().load(job_id).await
84    }
85
86    async fn save(&self, state: &JobState) -> Result<(), Self::Error> {
87        self.as_ref().save(state).await
88    }
89
90    async fn delete(&self, job_id: &str) -> Result<(), Self::Error> {
91        self.as_ref().delete(job_id).await
92    }
93
94    async fn drain_events(&self) -> Result<Vec<StoreEvent>, Self::Error> {
95        self.as_ref().drain_events().await
96    }
97
98    fn classify_error(error: &Self::Error) -> StoreErrorKind
99    where
100        Self: Sized,
101    {
102        T::classify_error(error)
103    }
104}
105
106impl ResilientStoreError for Infallible {
107    fn is_connection_issue(&self) -> bool {
108        match *self {}
109    }
110}
scheduler/store/mod.rs

scheduler/store/
mod.rs
