laminar_connectors/cdc/postgres/
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        (self.0 >> 32) as u32
45    }
46
47    /// Returns the lower 32 bits (offset within segment).
48    #[must_use]
49    #[allow(clippy::cast_possible_truncation)] // Intentional: extracts lower 32 bits of u64 LSN
50    pub const fn offset(self) -> u32 {
51        self.0 as u32
52    }
53
54    /// Returns the byte difference between two LSNs.
55    ///
56    /// Returns 0 if `other` is ahead of `self`.
57    #[must_use]
58    pub const fn diff(self, other: Lsn) -> u64 {
59        self.0.saturating_sub(other.0)
60    }
61
62    /// Advances the LSN by the given number of bytes.
63    #[must_use]
64    pub const fn advance(self, bytes: u64) -> Lsn {
65        Lsn(self.0.saturating_add(bytes))
66    }
67
68    /// Returns `true` if this is the zero LSN.
69    #[must_use]
70    pub const fn is_zero(self) -> bool {
71        self.0 == 0
72    }
73}
74
75impl fmt::Display for Lsn {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        write!(f, "{:X}/{:X}", self.segment(), self.offset())
78    }
79}
80
81impl FromStr for Lsn {
82    type Err = LsnParseError;
83
84    fn from_str(s: &str) -> Result<Self, Self::Err> {
85        let (high, low) = s
86            .split_once('/')
87            .ok_or_else(|| LsnParseError::InvalidFormat(s.to_string()))?;
88
89        let high = u32::from_str_radix(high, 16)
90            .map_err(|_| LsnParseError::InvalidHex(high.to_string()))?;
91        let low =
92            u32::from_str_radix(low, 16).map_err(|_| LsnParseError::InvalidHex(low.to_string()))?;
93
94        Ok(Lsn((u64::from(high) << 32) | u64::from(low)))
95    }
96}
97
98impl From<u64> for Lsn {
99    fn from(value: u64) -> Self {
100        Lsn(value)
101    }
102}
103
104impl From<Lsn> for u64 {
105    fn from(lsn: Lsn) -> Self {
106        lsn.0
107    }
108}
109
110/// Errors that can occur when parsing an LSN string.
111#[derive(Debug, Clone, thiserror::Error)]
112pub enum LsnParseError {
113    /// The string does not contain the expected `X/Y` format.
114    #[error("invalid LSN format (expected X/Y): {0}")]
115    InvalidFormat(String),
116
117    /// A hex component could not be parsed.
118    #[error("invalid hex in LSN: {0}")]
119    InvalidHex(String),
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125
126    #[test]
127    fn test_parse_valid_lsn() {
128        let lsn: Lsn = "0/1234ABCD".parse().unwrap();
129        assert_eq!(lsn.segment(), 0);
130        assert_eq!(lsn.offset(), 0x1234_ABCD);
131        assert_eq!(lsn.as_u64(), 0x0000_0000_1234_ABCD);
132    }
133
134    #[test]
135    fn test_parse_with_high_segment() {
136        let lsn: Lsn = "1/0".parse().unwrap();
137        assert_eq!(lsn.segment(), 1);
138        assert_eq!(lsn.offset(), 0);
139        assert_eq!(lsn.as_u64(), 0x0000_0001_0000_0000);
140    }
141
142    #[test]
143    fn test_parse_max_lsn() {
144        let lsn: Lsn = "FFFFFFFF/FFFFFFFF".parse().unwrap();
145        assert_eq!(lsn, Lsn::MAX);
146    }
147
148    #[test]
149    fn test_parse_invalid_no_slash() {
150        assert!("12345".parse::<Lsn>().is_err());
151    }
152
153    #[test]
154    fn test_parse_invalid_hex() {
155        assert!("ZZ/1234".parse::<Lsn>().is_err());
156        assert!("0/GHIJ".parse::<Lsn>().is_err());
157    }
158
159    #[test]
160    fn test_display() {
161        let lsn = Lsn::new(0x0000_0001_1234_ABCD);
162        assert_eq!(lsn.to_string(), "1/1234ABCD");
163    }
164
165    #[test]
166    fn test_display_zero() {
167        assert_eq!(Lsn::ZERO.to_string(), "0/0");
168    }
169
170    #[test]
171    fn test_roundtrip() {
172        let original = "A/BC1234";
173        let lsn: Lsn = original.parse().unwrap();
174        assert_eq!(lsn.to_string(), original);
175    }
176
177    #[test]
178    fn test_ordering() {
179        let a: Lsn = "0/100".parse().unwrap();
180        let b: Lsn = "0/200".parse().unwrap();
181        let c: Lsn = "1/0".parse().unwrap();
182        assert!(a < b);
183        assert!(b < c);
184        assert!(a < c);
185    }
186
187    #[test]
188    fn test_diff() {
189        let a: Lsn = "0/200".parse().unwrap();
190        let b: Lsn = "0/100".parse().unwrap();
191        assert_eq!(a.diff(b), 0x100);
192        assert_eq!(b.diff(a), 0); // saturating
193    }
194
195    #[test]
196    fn test_advance() {
197        let lsn: Lsn = "0/100".parse().unwrap();
198        let advanced = lsn.advance(256);
199        assert_eq!(advanced.to_string(), "0/200");
200    }
201
202    #[test]
203    fn test_is_zero() {
204        assert!(Lsn::ZERO.is_zero());
205        assert!(!Lsn::new(1).is_zero());
206    }
207
208    #[test]
209    fn test_from_u64() {
210        let lsn = Lsn::from(42u64);
211        assert_eq!(lsn.as_u64(), 42);
212        let val: u64 = lsn.into();
213        assert_eq!(val, 42);
214    }
215}
laminar_connectors/cdc/postgres/lsn.rs

laminar_connectors/cdc/postgres/
lsn.rs
