crabka_connect_postgres/
offset.rs

1use std::fmt;
2use std::str::FromStr;
3
4use crabka_connect::{OffsetMap, OffsetValue, SourceOffset};
5
6use crate::PostgresConnectError;
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
9pub struct PgLsn(pub u64);
10
11impl PgLsn {
12    #[must_use]
13    pub fn to_source_offset(self, database: &str, slot: &str) -> SourceOffset {
14        let partition = OffsetMap::from([
15            (
16                "database".to_owned(),
17                OffsetValue::String(database.to_owned()),
18            ),
19            ("slot".to_owned(), OffsetValue::String(slot.to_owned())),
20        ]);
21        let position = OffsetMap::from([("lsn".to_owned(), OffsetValue::String(self.to_string()))]);
22
23        SourceOffset::new(partition, position)
24    }
25
26    pub fn from_source_offset(
27        offset: &SourceOffset,
28        expected_slot: &str,
29    ) -> Result<PgLsn, PostgresConnectError> {
30        match offset.partition.get("slot") {
31            Some(OffsetValue::String(slot)) if slot == expected_slot => {}
32            Some(OffsetValue::String(slot)) => {
33                return Err(PostgresConnectError::Offset(format!(
34                    "source offset slot {slot:?} does not match expected slot {expected_slot:?}"
35                )));
36            }
37            _ => {
38                return Err(PostgresConnectError::Offset(
39                    "source offset missing string slot".to_owned(),
40                ));
41            }
42        }
43
44        match offset.position.get("lsn") {
45            Some(OffsetValue::String(lsn)) => lsn.parse(),
46            _ => Err(PostgresConnectError::Offset(
47                "source offset missing string lsn".to_owned(),
48            )),
49        }
50    }
51}
52
53impl fmt::Display for PgLsn {
54    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
55        let hi = self.0 >> 32;
56        let lo = self.0 & 0xffff_ffff;
57
58        write!(f, "{hi:X}/{lo:X}")
59    }
60}
61
62impl FromStr for PgLsn {
63    type Err = PostgresConnectError;
64
65    fn from_str(s: &str) -> Result<Self, Self::Err> {
66        let (hi, lo) = s
67            .split_once('/')
68            .ok_or_else(|| PostgresConnectError::Offset(format!("invalid Postgres LSN {s:?}")))?;
69
70        let hi = parse_lsn_half(hi, s)?;
71        let lo = parse_lsn_half(lo, s)?;
72
73        Ok(PgLsn((hi << 32) | lo))
74    }
75}
76
77fn parse_lsn_half(half: &str, lsn: &str) -> Result<u64, PostgresConnectError> {
78    let value = u64::from_str_radix(half, 16)
79        .map_err(|_| PostgresConnectError::Offset(format!("invalid Postgres LSN {lsn:?}")))?;
80
81    if value > 0xffff_ffff {
82        return Err(PostgresConnectError::Offset(format!(
83            "invalid Postgres LSN {lsn:?}: component exceeds 32 bits"
84        )));
85    }
86
87    Ok(value)
88}
89
90#[cfg(test)]
91mod tests {
92    use crabka_connect::{OffsetValue, SourceOffset};
93
94    use super::PgLsn;
95    use crate::PostgresConnectError;
96
97    #[test]
98    fn lsn_round_trips_postgres_text_form() {
99        let lsn: PgLsn = "16/B374D848".parse().expect("valid LSN");
100
101        assert_eq!(lsn, PgLsn(0x16_b374_d848));
102        assert_eq!(lsn.to_string(), "16/B374D848");
103        assert_eq!(
104            "FFFFFFFF/FFFFFFFF".parse::<PgLsn>().expect("max LSN"),
105            PgLsn(u64::MAX)
106        );
107    }
108
109    #[test]
110    fn source_offset_round_trips_and_checks_slot() {
111        let lsn: PgLsn = "0/2A".parse().expect("valid LSN");
112        let offset = lsn.to_source_offset("app", "slot_a");
113
114        assert_eq!(
115            PgLsn::from_source_offset(&offset, "slot_a").expect("matching slot"),
116            lsn
117        );
118        match PgLsn::from_source_offset(&offset, "slot_b").expect_err("slot mismatch") {
119            PostgresConnectError::Offset(message) => {
120                assert!(message.contains("does not match expected slot"));
121                assert!(message.contains("slot_a"));
122                assert!(message.contains("slot_b"));
123            }
124            error => panic!("expected offset error, got {error:?}"),
125        }
126    }
127
128    #[test]
129    fn source_offset_rejects_missing_or_non_string_slot() {
130        let missing = SourceOffset::default();
131        let mut non_string = PgLsn(42).to_source_offset("app", "slot_a");
132        non_string
133            .partition
134            .insert("slot".to_owned(), OffsetValue::Long(7));
135
136        for offset in [missing, non_string] {
137            match PgLsn::from_source_offset(&offset, "slot_a").expect_err("slot should fail") {
138                PostgresConnectError::Offset(message) => {
139                    assert_eq!(message, "source offset missing string slot");
140                }
141                error => panic!("expected offset error, got {error:?}"),
142            }
143        }
144    }
145
146    #[test]
147    fn malformed_lsn_is_offset_error() {
148        for input in ["not-a-lsn", "1/XYZ", "100000000/0"] {
149            assert!(matches!(
150                input.parse::<PgLsn>(),
151                Err(PostgresConnectError::Offset(_))
152            ));
153        }
154    }
155}
crabka_connect_postgres/offset.rs

crabka_connect_postgres/
offset.rs
