laminar_connectors/postgres/cdc/
lsn.rs

1//! `PostgreSQL` Log Sequence Number (LSN) type.
2//!
3//! An LSN is a 64-bit integer representing a byte position in the WAL stream.
4//! `PostgreSQL` displays LSNs in the format `X/Y` where X is the upper 32 bits
5//! and Y is the lower 32 bits, both in hexadecimal.
6
7use std::fmt;
8use std::str::FromStr;
9
10/// A `PostgreSQL` Log Sequence Number (LSN).
11///
12/// Represents a byte offset in the write-ahead log. Used to track
13/// replication progress and checkpoint positions.
14///
15/// # Format
16///
17/// LSNs are displayed as `X/YYYYYYYY` where X and Y are hex values.
18/// For example: `0/1234ABCD`, `1/0`, `FF/FFFFFFFF`.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
20pub struct Lsn(u64);
21
22impl Lsn {
23    /// The zero LSN, representing the start of the WAL.
24    pub const ZERO: Lsn = Lsn(0);
25
26    /// The maximum possible LSN.
27    pub const MAX: Lsn = Lsn(u64::MAX);
28
29    /// Creates a new LSN from a raw 64-bit value.
30    #[must_use]
31    pub const fn new(value: u64) -> Self {
32        Lsn(value)
33    }
34
35    /// Returns the raw 64-bit value.
36    #[must_use]
37    pub const fn as_u64(self) -> u64 {
38        self.0
39    }
40
41    /// Returns the upper 32 bits (segment number).
42    #[must_use]
43    pub const fn segment(self) -> u32 {
44        let bytes = self.0.to_be_bytes();
45        u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])
46    }
47
48    /// Returns the lower 32 bits (offset within segment).
49    #[must_use]
50    pub const fn offset(self) -> u32 {
51        let bytes = self.0.to_be_bytes();
52        u32::from_be_bytes([bytes[4], bytes[5], bytes[6], bytes[7]])
53    }
54
55    /// Returns the byte difference between two LSNs.
56    ///
57    /// Returns 0 if `other` is ahead of `self`.
58    #[must_use]
59    pub const fn diff(self, other: Lsn) -> u64 {
60        self.0.saturating_sub(other.0)
61    }
62
63    /// Advances the LSN by the given number of bytes.
64    #[must_use]
65    pub const fn advance(self, bytes: u64) -> Lsn {
66        Lsn(self.0.saturating_add(bytes))
67    }
68
69    /// Returns `true` if this is the zero LSN.
70    #[must_use]
71    pub const fn is_zero(self) -> bool {
72        self.0 == 0
73    }
74}
75
76impl fmt::Display for Lsn {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        write!(f, "{:X}/{:X}", self.segment(), self.offset())
79    }
80}
81
82impl FromStr for Lsn {
83    type Err = LsnParseError;
84
85    fn from_str(s: &str) -> Result<Self, Self::Err> {
86        let (high, low) = s
87            .split_once('/')
88            .ok_or_else(|| LsnParseError::InvalidFormat(s.to_string()))?;
89
90        let high = u32::from_str_radix(high, 16)
91            .map_err(|_| LsnParseError::InvalidHex(high.to_string()))?;
92        let low =
93            u32::from_str_radix(low, 16).map_err(|_| LsnParseError::InvalidHex(low.to_string()))?;
94
95        Ok(Lsn((u64::from(high) << 32) | u64::from(low)))
96    }
97}
98
99impl From<u64> for Lsn {
100    fn from(value: u64) -> Self {
101        Lsn(value)
102    }
103}
104
105impl From<Lsn> for u64 {
106    fn from(lsn: Lsn) -> Self {
107        lsn.0
108    }
109}
110
111/// Errors that can occur when parsing an LSN string.
112#[derive(Debug, Clone, thiserror::Error)]
113pub enum LsnParseError {
114    /// The string does not contain the expected `X/Y` format.
115    #[error("invalid LSN format (expected X/Y): {0}")]
116    InvalidFormat(String),
117
118    /// A hex component could not be parsed.
119    #[error("invalid hex in LSN: {0}")]
120    InvalidHex(String),
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126
127    #[test]
128    fn test_parse_valid_lsn() {
129        let lsn: Lsn = "0/1234ABCD".parse().unwrap();
130        assert_eq!(lsn.segment(), 0);
131        assert_eq!(lsn.offset(), 0x1234_ABCD);
132        assert_eq!(lsn.as_u64(), 0x0000_0000_1234_ABCD);
133    }
134
135    #[test]
136    fn test_parse_with_high_segment() {
137        let lsn: Lsn = "1/0".parse().unwrap();
138        assert_eq!(lsn.segment(), 1);
139        assert_eq!(lsn.offset(), 0);
140        assert_eq!(lsn.as_u64(), 0x0000_0001_0000_0000);
141    }
142
143    #[test]
144    fn test_parse_max_lsn() {
145        let lsn: Lsn = "FFFFFFFF/FFFFFFFF".parse().unwrap();
146        assert_eq!(lsn, Lsn::MAX);
147    }
148
149    #[test]
150    fn test_parse_invalid_no_slash() {
151        assert!("12345".parse::<Lsn>().is_err());
152    }
153
154    #[test]
155    fn test_parse_invalid_hex() {
156        assert!("ZZ/1234".parse::<Lsn>().is_err());
157        assert!("0/GHIJ".parse::<Lsn>().is_err());
158    }
159
160    #[test]
161    fn test_display() {
162        let lsn = Lsn::new(0x0000_0001_1234_ABCD);
163        assert_eq!(lsn.to_string(), "1/1234ABCD");
164    }
165
166    #[test]
167    fn test_display_zero() {
168        assert_eq!(Lsn::ZERO.to_string(), "0/0");
169    }
170
171    #[test]
172    fn test_roundtrip() {
173        let original = "A/BC1234";
174        let lsn: Lsn = original.parse().unwrap();
175        assert_eq!(lsn.to_string(), original);
176    }
177
178    #[test]
179    fn test_ordering() {
180        let a: Lsn = "0/100".parse().unwrap();
181        let b: Lsn = "0/200".parse().unwrap();
182        let c: Lsn = "1/0".parse().unwrap();
183        assert!(a < b);
184        assert!(b < c);
185        assert!(a < c);
186    }
187
188    #[test]
189    fn test_diff() {
190        let a: Lsn = "0/200".parse().unwrap();
191        let b: Lsn = "0/100".parse().unwrap();
192        assert_eq!(a.diff(b), 0x100);
193        assert_eq!(b.diff(a), 0); // saturating
194    }
195
196    #[test]
197    fn test_advance() {
198        let lsn: Lsn = "0/100".parse().unwrap();
199        let advanced = lsn.advance(256);
200        assert_eq!(advanced.to_string(), "0/200");
201    }
202
203    #[test]
204    fn test_is_zero() {
205        assert!(Lsn::ZERO.is_zero());
206        assert!(!Lsn::new(1).is_zero());
207    }
208
209    #[test]
210    fn test_from_u64() {
211        let lsn = Lsn::from(42u64);
212        assert_eq!(lsn.as_u64(), 42);
213        let val: u64 = lsn.into();
214        assert_eq!(val, 42);
215    }
216}
laminar_connectors/postgres/cdc/lsn.rs

laminar_connectors/postgres/cdc/
lsn.rs
