faucet_source_mysql_cdc/
state.rs

1//! State key + bookmark serialization for MySQL binlog progress.
2//!
3//! Bookmark JSON shapes:
4//! ```json
5//! { "file": "binlog.000123", "pos": 4567 }
6//! { "gtid_set": "uuid:1-1000" }
7//! ```
8
9use faucet_core::FaucetError;
10use serde::{Deserialize, Serialize};
11use serde_json::Value;
12
13/// State key for a given replica `server_id`.
14pub fn state_key(server_id: u32) -> String {
15    format!("mysql-cdc:{server_id}")
16}
17
18/// Durable bookmark: either a binlog file+position or an executed GTID set.
19#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
20#[serde(untagged)]
21pub enum Bookmark {
22    /// Binlog coordinates.
23    FilePos { file: String, pos: u64 },
24    /// Executed GTID set (e.g. `uuid:1-1000`).
25    GtidSet { gtid_set: String },
26}
27
28impl Bookmark {
29    /// Parse a bookmark previously emitted by `to_value`.
30    pub fn from_value(v: Value) -> Result<Self, FaucetError> {
31        serde_json::from_value(v)
32            .map_err(|e| FaucetError::State(format!("mysql-cdc bookmark parse: {e}")))
33    }
34
35    /// Serialize for the state store.
36    pub fn to_value(&self) -> Result<Value, FaucetError> {
37        serde_json::to_value(self)
38            .map_err(|e| FaucetError::State(format!("mysql-cdc bookmark serialize: {e}")))
39    }
40}
41
42#[cfg(test)]
43mod tests {
44    use super::*;
45    use serde_json::json;
46
47    #[test]
48    fn state_key_format() {
49        assert_eq!(state_key(1001), "mysql-cdc:1001");
50    }
51
52    #[test]
53    fn file_pos_round_trip() {
54        let b = Bookmark::FilePos {
55            file: "binlog.000123".into(),
56            pos: 4567,
57        };
58        assert_eq!(Bookmark::from_value(b.to_value().unwrap()).unwrap(), b);
59    }
60
61    #[test]
62    fn gtid_round_trip() {
63        let b = Bookmark::GtidSet {
64            gtid_set: "uuid:1-1000".into(),
65        };
66        assert_eq!(Bookmark::from_value(b.to_value().unwrap()).unwrap(), b);
67    }
68
69    #[test]
70    fn file_pos_json_shape() {
71        let v = Bookmark::FilePos {
72            file: "b.1".into(),
73            pos: 9,
74        }
75        .to_value()
76        .unwrap();
77        assert_eq!(v, json!({ "file": "b.1", "pos": 9 }));
78    }
79
80    #[test]
81    fn rejects_garbage() {
82        assert!(Bookmark::from_value(json!({ "nope": 1 })).is_err());
83    }
84}
faucet_source_mysql_cdc/state.rs

faucet_source_mysql_cdc/
state.rs
