pgsync_core/
checkpoint.rs

1//! Durable replication progress.
2
3use crate::lsn::Lsn;
4use crate::state::JobState;
5use serde::{Deserialize, Serialize};
6
7/// Everything needed to identify a job's source and resume it safely.
8///
9/// Invariants maintained by the engine:
10/// - `commit_lsn <= committed_lsn <= acked_lsn`
11/// - the slot's `confirmed_flush_lsn` never exceeds `acked_lsn`
12/// - `committed_lsn` only moves inside the target transaction that applied the data
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14pub struct Checkpoint {
15    pub job_id: String,
16    pub state: JobState,
17    /// `IDENTIFY_SYSTEM.systemid` of the source cluster.
18    pub source_system_id: String,
19    pub source_dbname: String,
20    pub slot: String,
21    pub publication: String,
22    /// Whether PGSync created the slot (and may therefore drop it in `cleanup`).
23    pub slot_owned: bool,
24    /// `consistent_point` returned at slot creation.
25    pub consistent_point: Option<Lsn>,
26    /// `commit_lsn` of the last applied transaction (`Begin.final_lsn` of a redelivered
27    /// transaction is compared against this).
28    pub commit_lsn: Lsn,
29    /// `end_lsn` of the last applied transaction.
30    pub committed_lsn: Lsn,
31    /// Highest position reported to the server as flushed.
32    pub acked_lsn: Lsn,
33    pub last_xid: Option<u32>,
34    /// PostgreSQL timestamp (µs since 2000-01-01) of the last applied commit.
35    pub last_commit_ts: Option<i64>,
36    pub txns_applied: u64,
37    pub last_error: Option<String>,
38}
39
40impl Checkpoint {
41    pub fn new_job(
42        job_id: &str,
43        source_system_id: &str,
44        source_dbname: &str,
45        slot: &str,
46        publication: &str,
47    ) -> Self {
48        Checkpoint {
49            job_id: job_id.to_owned(),
50            state: JobState::Initializing,
51            source_system_id: source_system_id.to_owned(),
52            source_dbname: source_dbname.to_owned(),
53            slot: slot.to_owned(),
54            publication: publication.to_owned(),
55            slot_owned: false,
56            consistent_point: None,
57            commit_lsn: Lsn::ZERO,
58            committed_lsn: Lsn::ZERO,
59            acked_lsn: Lsn::ZERO,
60            last_xid: None,
61            last_commit_ts: None,
62            txns_applied: 0,
63            last_error: None,
64        }
65    }
66
67    /// True when a transaction whose commit record is at `final_lsn` has already been
68    /// applied and must be skipped.
69    pub fn already_applied(&self, final_lsn: Lsn) -> bool {
70        final_lsn <= self.commit_lsn
71    }
72
73    /// Record a transaction commit. Keeps the invariants.
74    pub fn record_commit(&mut self, commit_lsn: Lsn, end_lsn: Lsn, xid: u32, commit_ts: i64) {
75        debug_assert!(commit_lsn > self.commit_lsn);
76        self.commit_lsn = commit_lsn;
77        self.committed_lsn = self.committed_lsn.max(end_lsn);
78        self.acked_lsn = self.acked_lsn.max(self.committed_lsn);
79        self.last_xid = Some(xid);
80        self.last_commit_ts = Some(commit_ts);
81        self.txns_applied += 1;
82    }
83
84    /// Record a keepalive position as safe to acknowledge. Returns true if it advanced.
85    pub fn record_keepalive(&mut self, wal_end: Lsn) -> bool {
86        if wal_end > self.acked_lsn {
87            self.acked_lsn = wal_end;
88            true
89        } else {
90            false
91        }
92    }
93
94    pub fn check_invariants(&self) -> Result<(), String> {
95        if self.commit_lsn > self.committed_lsn {
96            return Err(format!(
97                "commit_lsn {} > committed_lsn {}",
98                self.commit_lsn, self.committed_lsn
99            ));
100        }
101        if self.committed_lsn > self.acked_lsn {
102            return Err(format!(
103                "committed_lsn {} > acked_lsn {}",
104                self.committed_lsn, self.acked_lsn
105            ));
106        }
107        Ok(())
108    }
109}
110
111/// Persists [`Checkpoint`]s. The PostgreSQL implementation stores them in the target.
112pub trait CheckpointStore: Send {
113    fn load(
114        &mut self,
115        job_id: &str,
116    ) -> impl Future<Output = crate::Result<Option<Checkpoint>>> + Send;
117    fn save(&mut self, cp: &Checkpoint) -> impl Future<Output = crate::Result<()>> + Send;
118}
119
120#[cfg(test)]
121mod tests {
122    use super::*;
123
124    fn cp() -> Checkpoint {
125        Checkpoint::new_job("j", "sys", "db", "slot", "pub")
126    }
127
128    #[test]
129    fn commit_advances_all_positions() {
130        let mut c = cp();
131        c.record_commit(Lsn(100), Lsn(120), 7, 1);
132        assert_eq!(c.commit_lsn, Lsn(100));
133        assert_eq!(c.committed_lsn, Lsn(120));
134        assert_eq!(c.acked_lsn, Lsn(120));
135        assert_eq!(c.txns_applied, 1);
136        c.check_invariants().unwrap();
137        assert!(c.already_applied(Lsn(100)));
138        assert!(c.already_applied(Lsn(50)));
139        assert!(!c.already_applied(Lsn(101)));
140    }
141
142    #[test]
143    fn keepalive_only_moves_ack_forward() {
144        let mut c = cp();
145        c.record_commit(Lsn(100), Lsn(120), 7, 1);
146        assert!(c.record_keepalive(Lsn(500)));
147        assert_eq!(c.acked_lsn, Lsn(500));
148        assert!(!c.record_keepalive(Lsn(400)));
149        assert_eq!(c.committed_lsn, Lsn(120));
150        c.check_invariants().unwrap();
151        // A later commit below the acked position keeps acked where it is.
152        c.record_commit(Lsn(300), Lsn(320), 8, 2);
153        assert_eq!(c.acked_lsn, Lsn(500));
154        c.check_invariants().unwrap();
155    }
156
157    #[test]
158    fn serde_round_trip() {
159        let mut c = cp();
160        c.record_commit(Lsn(0x1_0000_0010), Lsn(0x1_0000_0020), 9, 123);
161        let json = serde_json::to_string(&c).unwrap();
162        assert!(json.contains("\"1/10\""));
163        let back: Checkpoint = serde_json::from_str(&json).unwrap();
164        assert_eq!(back, c);
165    }
166}
pgsync_core/checkpoint.rs

pgsync_core/
checkpoint.rs
