faucet_source_mssql_cdc/
lsn.rs

1//! SQL Server Log Sequence Number (LSN) — pure parse / format / compare logic.
2//!
3//! An LSN in SQL Server change data capture is a `binary(10)` value: a 10-byte,
4//! big-endian, fixed-width integer whose lexicographic byte order is exactly its
5//! numeric order. We keep it as `[u8; 10]` and serialise it to the state store as
6//! a 20-character lowercase hex string (matching `CONVERT(VARCHAR(20), lsn, 2)`).
7//!
8//! All logic here is pure and unit-tested without a live server — the LSN is
9//! load-bearing for resumability, so a subtle off-by-one in the increment or a
10//! wrong comparison would silently drop or replay change rows.
11
12use std::fmt;
13
14use faucet_core::FaucetError;
15
16/// Number of bytes in a SQL Server `binary(10)` LSN.
17pub const LSN_BYTES: usize = 10;
18
19/// A SQL Server change data capture Log Sequence Number.
20///
21/// Stored as a big-endian byte array; [`Ord`] is the derived lexicographic
22/// comparison, which for big-endian bytes equals numeric order.
23#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
24pub struct Lsn([u8; LSN_BYTES]);
25
26impl Lsn {
27    /// The all-zero LSN (`0x0000...0000`), the minimum possible value.
28    pub const ZERO: Lsn = Lsn([0u8; LSN_BYTES]);
29
30    /// Wrap a raw 10-byte big-endian array.
31    pub fn from_array(bytes: [u8; LSN_BYTES]) -> Self {
32        Lsn(bytes)
33    }
34
35    /// Build an LSN from a byte slice. The slice must be exactly `LSN_BYTES`
36    /// long; any other length is a typed error (SQL Server always returns
37    /// `binary(10)`, so a different width means a decode bug, not a data value).
38    pub fn from_bytes(bytes: &[u8]) -> Result<Self, FaucetError> {
39        if bytes.len() != LSN_BYTES {
40            return Err(FaucetError::Source(format!(
41                "mssql-cdc: LSN must be exactly {LSN_BYTES} bytes, got {}",
42                bytes.len()
43            )));
44        }
45        let mut arr = [0u8; LSN_BYTES];
46        arr.copy_from_slice(bytes);
47        Ok(Lsn(arr))
48    }
49
50    /// Parse an LSN from its hex string form (20 hex digits, case-insensitive, an
51    /// optional `0x`/`0X` prefix tolerated). Rejects any other length or a
52    /// non-hex digit with a typed error.
53    pub fn from_hex(s: &str) -> Result<Self, FaucetError> {
54        let hex = s
55            .strip_prefix("0x")
56            .or_else(|| s.strip_prefix("0X"))
57            .unwrap_or(s);
58        if hex.len() != LSN_BYTES * 2 {
59            return Err(FaucetError::Source(format!(
60                "mssql-cdc: LSN hex must be {} digits, got {} ({s:?})",
61                LSN_BYTES * 2,
62                hex.len()
63            )));
64        }
65        let mut arr = [0u8; LSN_BYTES];
66        for (i, byte) in arr.iter_mut().enumerate() {
67            let pair = &hex[i * 2..i * 2 + 2];
68            *byte = u8::from_str_radix(pair, 16).map_err(|e| {
69                FaucetError::Source(format!("mssql-cdc: invalid LSN hex {s:?}: {e}"))
70            })?;
71        }
72        Ok(Lsn(arr))
73    }
74
75    /// Render as a 20-character lowercase hex string (no `0x` prefix), matching
76    /// `CONVERT(VARCHAR(20), lsn, 2)`.
77    pub fn to_hex(&self) -> String {
78        let mut out = String::with_capacity(LSN_BYTES * 2);
79        for b in &self.0 {
80            out.push_str(&format!("{b:02x}"));
81        }
82        out
83    }
84
85    /// The smallest LSN strictly greater than `self` — the client-side analogue
86    /// of `sys.fn_cdc_increment_lsn`. Used to derive the **half-open** lower
87    /// bound for the next poll from a durable bookmark, so a resumed run never
88    /// re-reads the last already-committed change (the `fn_cdc_get_all_changes`
89    /// range is inclusive of `from_lsn`).
90    ///
91    /// Returns `None` only when `self` is the all-`0xFF` maximum (unreachable in
92    /// practice — the log would have to wrap the entire 80-bit LSN space).
93    pub fn increment(&self) -> Option<Self> {
94        let mut arr = self.0;
95        for byte in arr.iter_mut().rev() {
96            if *byte == u8::MAX {
97                *byte = 0;
98            } else {
99                *byte += 1;
100                return Some(Lsn(arr));
101            }
102        }
103        None
104    }
105}
106
107impl fmt::Debug for Lsn {
108    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
109        write!(f, "Lsn({})", self.to_hex())
110    }
111}
112
113impl fmt::Display for Lsn {
114    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
115        f.write_str(&self.to_hex())
116    }
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122
123    #[test]
124    fn hex_round_trips() {
125        let hex = "0000002a000000550003";
126        let lsn = Lsn::from_hex(hex).unwrap();
127        assert_eq!(lsn.to_hex(), hex);
128    }
129
130    #[test]
131    fn hex_tolerates_0x_prefix_and_uppercase() {
132        let a = Lsn::from_hex("0x0000002A000000550003").unwrap();
133        let b = Lsn::from_hex("0000002a000000550003").unwrap();
134        assert_eq!(a, b);
135    }
136
137    #[test]
138    fn hex_rejects_wrong_length_and_non_hex() {
139        assert!(Lsn::from_hex("00").is_err());
140        assert!(Lsn::from_hex("").is_err());
141        // 20 chars but a non-hex digit.
142        assert!(Lsn::from_hex("0000002a000000550zzz").is_err());
143    }
144
145    #[test]
146    fn from_bytes_requires_exact_width() {
147        assert!(Lsn::from_bytes(&[0u8; 10]).is_ok());
148        assert!(Lsn::from_bytes(&[0u8; 9]).is_err());
149        assert!(Lsn::from_bytes(&[0u8; 11]).is_err());
150    }
151
152    #[test]
153    fn ordering_is_numeric_big_endian() {
154        let small = Lsn::from_hex("00000000000000000001").unwrap();
155        let big = Lsn::from_hex("00000000000000000002").unwrap();
156        let bigger_high_byte = Lsn::from_hex("01000000000000000000").unwrap();
157        assert!(small < big);
158        assert!(big < bigger_high_byte);
159        assert_eq!(Lsn::ZERO, Lsn::from_hex("00000000000000000000").unwrap());
160        assert!(Lsn::ZERO < small);
161    }
162
163    #[test]
164    fn increment_is_the_next_value() {
165        let lsn = Lsn::from_hex("000000000000000000ff").unwrap();
166        let next = lsn.increment().unwrap();
167        // 0x...00ff + 1 carries into the next byte -> 0x...0100.
168        assert_eq!(next.to_hex(), "00000000000000000100");
169        assert!(lsn < next);
170        // No LSN sits strictly between lsn and its increment.
171        assert!(Lsn::from_hex("00000000000000000100").unwrap() <= next);
172    }
173
174    #[test]
175    fn increment_simple_low_byte() {
176        let lsn = Lsn::from_hex("00000000000000000001").unwrap();
177        assert_eq!(lsn.increment().unwrap().to_hex(), "00000000000000000002");
178    }
179
180    #[test]
181    fn increment_max_saturates_to_none() {
182        let max = Lsn::from_array([0xFF; LSN_BYTES]);
183        assert!(max.increment().is_none());
184    }
185
186    #[test]
187    fn increment_of_zero_is_one() {
188        assert_eq!(
189            Lsn::ZERO.increment().unwrap().to_hex(),
190            "00000000000000000001"
191        );
192    }
193}
faucet_source_mssql_cdc/lsn.rs

faucet_source_mssql_cdc/
lsn.rs
