faucet_source_mssql_cdc/
state.rs

1//! Durable bookmark for SQL Server CDC progress.
2//!
3//! Because one source may poll several capture instances, the bookmark is a
4//! **map** of capture-instance name → last-committed LSN (hex). Storing the
5//! whole map in every emitted page's bookmark keeps the pipeline's single
6//! state-key model intact: each page persists the complete, up-to-date map.
7//!
8//! JSON shape:
9//! ```json
10//! { "dbo_Orders": "0000002a000000550003", "dbo_Items": "0000002a000000560001" }
11//! ```
12
13use std::collections::BTreeMap;
14
15use faucet_core::FaucetError;
16use serde_json::Value;
17
18use crate::lsn::Lsn;
19
20/// Per-capture-instance LSN bookmark map. Ordered (`BTreeMap`) so serialization
21/// is deterministic.
22#[derive(Clone, Debug, Default, PartialEq, Eq)]
23pub struct Bookmarks(BTreeMap<String, Lsn>);
24
25impl Bookmarks {
26    /// An empty bookmark map (fresh run, nothing committed yet).
27    pub fn new() -> Self {
28        Self(BTreeMap::new())
29    }
30
31    /// The last committed LSN for `capture_instance`, if any.
32    pub fn get(&self, capture_instance: &str) -> Option<Lsn> {
33        self.0.get(capture_instance).copied()
34    }
35
36    /// Record `lsn` as the last committed LSN for `capture_instance`.
37    pub fn set(&mut self, capture_instance: impl Into<String>, lsn: Lsn) {
38        self.0.insert(capture_instance.into(), lsn);
39    }
40
41    /// Parse a bookmark map previously produced by [`to_value`](Self::to_value).
42    ///
43    /// The `null`/absent case yields an empty map (a fresh run). Every value must
44    /// be a valid LSN hex string, else a typed [`FaucetError::State`].
45    pub fn from_value(v: Value) -> Result<Self, FaucetError> {
46        match v {
47            Value::Null => Ok(Self::new()),
48            Value::Object(map) => {
49                let mut out = BTreeMap::new();
50                for (ci, lsn_val) in map {
51                    let hex = lsn_val.as_str().ok_or_else(|| {
52                        FaucetError::State(format!(
53                            "mssql-cdc bookmark: value for {ci:?} must be an LSN hex string"
54                        ))
55                    })?;
56                    let lsn = Lsn::from_hex(hex)
57                        .map_err(|e| FaucetError::State(format!("mssql-cdc bookmark: {e}")))?;
58                    out.insert(ci, lsn);
59                }
60                Ok(Self(out))
61            }
62            other => Err(FaucetError::State(format!(
63                "mssql-cdc bookmark must be a JSON object of capture_instance -> LSN hex, got {other}"
64            ))),
65        }
66    }
67
68    /// Serialize the map for the state store.
69    pub fn to_value(&self) -> Result<Value, FaucetError> {
70        let mut map = serde_json::Map::with_capacity(self.0.len());
71        for (ci, lsn) in &self.0 {
72            map.insert(ci.clone(), Value::String(lsn.to_hex()));
73        }
74        Ok(Value::Object(map))
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81    use serde_json::json;
82
83    fn lsn(hex: &str) -> Lsn {
84        Lsn::from_hex(hex).unwrap()
85    }
86
87    #[test]
88    fn set_get_round_trip() {
89        let mut b = Bookmarks::new();
90        assert!(b.get("dbo_Orders").is_none());
91        b.set("dbo_Orders", lsn("00000000000000000005"));
92        assert_eq!(b.get("dbo_Orders"), Some(lsn("00000000000000000005")));
93    }
94
95    #[test]
96    fn value_round_trip() {
97        let mut b = Bookmarks::new();
98        b.set("dbo_Orders", lsn("0000002a000000550003"));
99        b.set("dbo_Items", lsn("0000002a000000560001"));
100        let v = b.to_value().unwrap();
101        assert_eq!(Bookmarks::from_value(v).unwrap(), b);
102    }
103
104    #[test]
105    fn value_shape_is_ci_to_hex() {
106        let mut b = Bookmarks::new();
107        b.set("dbo_Orders", lsn("00000000000000000009"));
108        assert_eq!(
109            b.to_value().unwrap(),
110            json!({ "dbo_Orders": "00000000000000000009" })
111        );
112    }
113
114    #[test]
115    fn null_is_empty_map() {
116        assert_eq!(
117            Bookmarks::from_value(Value::Null).unwrap(),
118            Bookmarks::new()
119        );
120    }
121
122    #[test]
123    fn rejects_non_object() {
124        assert!(Bookmarks::from_value(json!("nope")).is_err());
125        assert!(Bookmarks::from_value(json!([1, 2, 3])).is_err());
126    }
127
128    #[test]
129    fn rejects_bad_lsn_value() {
130        assert!(Bookmarks::from_value(json!({ "dbo_Orders": 42 })).is_err());
131        assert!(Bookmarks::from_value(json!({ "dbo_Orders": "xx" })).is_err());
132    }
133}
faucet_source_mssql_cdc/state.rs

faucet_source_mssql_cdc/
state.rs
