datum_cdc/
lsn.rs

1use std::{fmt, str::FromStr};
2
3use serde::{Deserialize, Serialize};
4
5use crate::{CdcError, CdcResult};
6
7/// PostgreSQL log sequence number.
8///
9/// PostgreSQL displays LSNs as `X/Y`, where `X` is the high 32 bits and `Y` is
10/// the low 32 bits in hexadecimal.
11#[derive(
12    Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
13)]
14pub struct PgLsn(u64);
15
16impl PgLsn {
17    /// Zero LSN. Passing this to PostgreSQL asks the slot to start from its
18    /// confirmed position.
19    pub const ZERO: Self = Self(0);
20
21    #[must_use]
22    pub const fn from_u64(value: u64) -> Self {
23        Self(value)
24    }
25
26    #[must_use]
27    pub const fn as_u64(self) -> u64 {
28        self.0
29    }
30
31    #[must_use]
32    pub const fn is_zero(self) -> bool {
33        self.0 == 0
34    }
35
36    pub fn parse(value: &str) -> CdcResult<Self> {
37        let Some((high, low)) = value.split_once('/') else {
38            return Err(CdcError::Config(format!(
39                "invalid LSN {value:?}: missing '/'"
40            )));
41        };
42        let high = u64::from_str_radix(high, 16)
43            .map_err(|err| CdcError::Config(format!("invalid LSN high half {high:?}: {err}")))?;
44        let low = u64::from_str_radix(low, 16)
45            .map_err(|err| CdcError::Config(format!("invalid LSN low half {low:?}: {err}")))?;
46        Ok(Self((high << 32) | low))
47    }
48}
49
50impl fmt::Display for PgLsn {
51    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
52        let high = (self.0 >> 32) as u32;
53        let low = (self.0 & 0xffff_ffff) as u32;
54        write!(f, "{high:X}/{low:X}")
55    }
56}
57
58impl From<u64> for PgLsn {
59    fn from(value: u64) -> Self {
60        Self(value)
61    }
62}
63
64impl From<PgLsn> for u64 {
65    fn from(value: PgLsn) -> Self {
66        value.as_u64()
67    }
68}
69
70impl From<pgwire_replication::Lsn> for PgLsn {
71    fn from(value: pgwire_replication::Lsn) -> Self {
72        Self(value.as_u64())
73    }
74}
75
76impl From<PgLsn> for pgwire_replication::Lsn {
77    fn from(value: PgLsn) -> Self {
78        pgwire_replication::Lsn::from_u64(value.as_u64())
79    }
80}
81
82impl FromStr for PgLsn {
83    type Err = CdcError;
84
85    fn from_str(value: &str) -> Result<Self, Self::Err> {
86        Self::parse(value)
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93
94    #[test]
95    fn formats_lsn_like_postgres() {
96        let lsn = PgLsn::from_u64(0x16_0000_0000 + 0xb374_d848);
97        assert_eq!(lsn.to_string(), "16/B374D848");
98        assert_eq!(PgLsn::parse("16/B374D848").unwrap(), lsn);
99    }
100}
datum_cdc/lsn.rs

datum_cdc/
lsn.rs
