pgsync_core/
state.rs

1//! Job lifecycle states.
2
3use serde::{Deserialize, Serialize};
4use std::fmt;
5
6/// Observable state of a replication job.
7#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
9pub enum JobState {
10    Initializing,
11    Snapshotting,
12    CatchingUp,
13    Streaming,
14    Paused,
15    Error,
16    ResyncRequired,
17}
18
19impl JobState {
20    pub fn as_str(self) -> &'static str {
21        match self {
22            JobState::Initializing => "INITIALIZING",
23            JobState::Snapshotting => "SNAPSHOTTING",
24            JobState::CatchingUp => "CATCHING_UP",
25            JobState::Streaming => "STREAMING",
26            JobState::Paused => "PAUSED",
27            JobState::Error => "ERROR",
28            JobState::ResyncRequired => "RESYNC_REQUIRED",
29        }
30    }
31
32    /// States that permit `pgsync run` to resume CDC.
33    pub fn is_resumable(self) -> bool {
34        matches!(
35            self,
36            JobState::CatchingUp | JobState::Streaming | JobState::Paused | JobState::Error
37        )
38    }
39}
40
41impl fmt::Display for JobState {
42    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43        f.write_str(self.as_str())
44    }
45}
46
47impl std::str::FromStr for JobState {
48    type Err = String;
49
50    fn from_str(s: &str) -> Result<Self, Self::Err> {
51        Ok(match s {
52            "INITIALIZING" => JobState::Initializing,
53            "SNAPSHOTTING" => JobState::Snapshotting,
54            "CATCHING_UP" => JobState::CatchingUp,
55            "STREAMING" => JobState::Streaming,
56            "PAUSED" => JobState::Paused,
57            "ERROR" => JobState::Error,
58            "RESYNC_REQUIRED" => JobState::ResyncRequired,
59            other => return Err(format!("unknown job state {other:?}")),
60        })
61    }
62}
63
64#[cfg(test)]
65mod tests {
66    use super::*;
67
68    #[test]
69    fn round_trips_through_text() {
70        for s in [
71            JobState::Initializing,
72            JobState::Snapshotting,
73            JobState::CatchingUp,
74            JobState::Streaming,
75            JobState::Paused,
76            JobState::Error,
77            JobState::ResyncRequired,
78        ] {
79            assert_eq!(s.as_str().parse::<JobState>().unwrap(), s);
80            assert_eq!(
81                serde_json::to_string(&s).unwrap(),
82                format!("\"{}\"", s.as_str())
83            );
84        }
85        assert!("STREAMIN".parse::<JobState>().is_err());
86    }
87
88    #[test]
89    fn resync_required_and_snapshotting_do_not_resume() {
90        assert!(!JobState::ResyncRequired.is_resumable());
91        assert!(!JobState::Snapshotting.is_resumable());
92        assert!(!JobState::Initializing.is_resumable());
93        assert!(JobState::Streaming.is_resumable());
94    }
95}
pgsync_core/state.rs

pgsync_core/
state.rs
