laminar_connectors/postgres/cdc/
postgres_io.rs

1//! `PostgreSQL` logical replication connections and slot administration.
2
3#[cfg(not(test))]
4use super::lsn::Lsn;
5#[cfg(not(test))]
6use crate::connector::ConnectorTaskGuard;
7use crate::error::ConnectorError;
8use sha2::{Digest, Sha256};
9
10pub(super) const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
11
12const MINIMUM_SERVER_VERSION_NUM: u32 = 170_000;
13
14/// Database-side identity that makes an engine checkpoint safe to resume.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub(super) struct PostgresCheckpointBinding {
17    pub system_identifier: u64,
18    pub timeline_id: u32,
19    pub database_oid: u32,
20    pub publication_oid: u32,
21    pub publication_definition_sha256: String,
22    pub source_config_sha256: String,
23    pub slot_plugin: String,
24    pub slot_two_phase: bool,
25    pub slot_failover: bool,
26}
27
28/// Read-only projection of the recovery fields on an existing slot.
29#[cfg(not(test))]
30pub(super) struct InspectedReplicationSlot {
31    pub confirmed_flush_lsn: Option<Lsn>,
32    pub binding: PostgresCheckpointBinding,
33}
34
35fn digest_field(digest: &mut Sha256, value: &[u8]) {
36    digest.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes());
37    digest.update(value);
38}
39
40/// Hashes only settings that change which logical changes Laminar emits.
41/// Connection endpoints and buffering limits deliberately remain restartable.
42#[must_use]
43pub(super) fn source_config_digest(config: &super::config::PostgresCdcConfig) -> String {
44    let mut digest = Sha256::new();
45    digest.update(b"laminardb-postgres-cdc-source-v1\0");
46    digest_field(&mut digest, b"pgoutput");
47    digest_field(&mut digest, b"proto_version=1");
48    digest_field(&mut digest, b"messages=false");
49
50    for tables in [&config.table_include, &config.table_exclude] {
51        let mut canonical: Vec<&str> = tables.iter().map(String::as_str).collect();
52        canonical.sort_unstable();
53        canonical.dedup();
54        digest.update(
55            u64::try_from(canonical.len())
56                .unwrap_or(u64::MAX)
57                .to_be_bytes(),
58        );
59        for table in canonical {
60            digest_field(&mut digest, table.as_bytes());
61        }
62    }
63
64    format!("{:x}", digest.finalize())
65}
66
67/// Cancellation-safe control-plane connection and driver task.
68#[cfg(not(test))]
69pub(super) struct ControlConnection {
70    client: tokio_postgres::Client,
71    handle: Option<tokio::task::JoinHandle<()>>,
72}
73
74#[cfg(not(test))]
75impl ControlConnection {
76    #[must_use]
77    pub(super) fn client(&self) -> &tokio_postgres::Client {
78        &self.client
79    }
80
81    pub(super) async fn close(mut self) {
82        if let Some(handle) = self.handle.take() {
83            handle.abort();
84            let _ = handle.await;
85        }
86    }
87}
88
89#[cfg(not(test))]
90impl Drop for ControlConnection {
91    fn drop(&mut self) {
92        if let Some(handle) = self.handle.take() {
93            handle.abort();
94        }
95    }
96}
97
98/// Opens the regular connection used for version and recovery-identity checks.
99///
100/// # Errors
101///
102/// Returns an error when TLS configuration is invalid or the connection cannot be opened.
103#[cfg(not(test))]
104pub(super) async fn connect(
105    config: &super::config::PostgresCdcConfig,
106    driver_guard: ConnectorTaskGuard,
107) -> Result<ControlConnection, ConnectorError> {
108    use crate::postgres::SslMode;
109
110    let pg_config = config.control_connection_config()?;
111    match config.ssl_mode {
112        SslMode::Disable => {
113            let (client, connection) =
114                tokio::time::timeout(CONNECT_TIMEOUT, pg_config.connect(tokio_postgres::NoTls))
115                    .await
116                    .map_err(|_| {
117                        ConnectorError::ConnectionFailed(
118                            "PostgreSQL connect timed out after 10 seconds".into(),
119                        )
120                    })?
121                    .map_err(|error| {
122                        ConnectorError::ConnectionFailed(format!("PostgreSQL connect: {error}"))
123                    })?;
124            let handle = tokio::spawn(async move {
125                let _driver_guard = driver_guard;
126                if let Err(error) = connection.await {
127                    tracing::error!(%error, "PostgreSQL control-plane connection error");
128                }
129            });
130            Ok(ControlConnection {
131                client,
132                handle: Some(handle),
133            })
134        }
135        SslMode::VerifyFull => {
136            let tls = crate::postgres::make_rustls_connector(config.ssl_ca_cert_path.as_deref())?;
137            let (client, connection) =
138                tokio::time::timeout(CONNECT_TIMEOUT, pg_config.connect(tls))
139                    .await
140                    .map_err(|_| {
141                        ConnectorError::ConnectionFailed(
142                            "PostgreSQL TLS connect timed out after 10 seconds".into(),
143                        )
144                    })?
145                    .map_err(|error| {
146                        ConnectorError::ConnectionFailed(format!("PostgreSQL TLS connect: {error}"))
147                    })?;
148            let handle = tokio::spawn(async move {
149                let _driver_guard = driver_guard;
150                if let Err(error) = connection.await {
151                    tracing::error!(%error, "PostgreSQL control-plane TLS connection error");
152                }
153            });
154            Ok(ControlConnection {
155                client,
156                handle: Some(handle),
157            })
158        }
159    }
160}
161
162/// Inspects an existing logical replication slot and returns its durable cursor.
163///
164/// This operation never creates, replaces, advances, or drops a slot. Recovery
165/// owns an exact engine checkpoint and must fail closed when the corresponding
166/// `PostgreSQL` slot is absent.
167///
168/// # Errors
169///
170/// Returns an error when slot lookup, identity validation, or LSN parsing fails.
171#[cfg(not(test))]
172pub(super) async fn inspect_replication_slot(
173    client: &tokio_postgres::Client,
174    slot_name: &str,
175    plugin: &str,
176    database: &str,
177    publication: &str,
178    source_config_sha256: String,
179) -> Result<Option<InspectedReplicationSlot>, ConnectorError> {
180    let version_row = tokio::time::timeout(
181        CONNECT_TIMEOUT,
182        client.query_one("SELECT current_setting('server_version_num')", &[]),
183    )
184    .await
185    .map_err(|_| {
186        ConnectorError::ConnectionFailed(
187            "query PostgreSQL server version timed out after 10 seconds".into(),
188        )
189    })?
190    .map_err(|error| {
191        ConnectorError::ConnectionFailed(format!("query PostgreSQL server version: {error}"))
192    })?;
193    let version_text: &str = version_row.get(0);
194    let version_num = version_text.parse::<u32>().map_err(|error| {
195        ConnectorError::ReadError(format!(
196            "invalid PostgreSQL server_version_num '{version_text}': {error}"
197        ))
198    })?;
199    validate_server_version_num(version_num)?;
200
201    let control_row = tokio::time::timeout(
202        CONNECT_TIMEOUT,
203        client.query_one(
204            "SELECT control_system.system_identifier::text, control_checkpoint.timeline_id::text \
205             FROM pg_catalog.pg_control_system() AS control_system \
206             CROSS JOIN pg_catalog.pg_control_checkpoint() AS control_checkpoint",
207            &[],
208        ),
209    )
210    .await
211    .map_err(|_| {
212        ConnectorError::ConnectionFailed(
213            "query PostgreSQL system identifier and timeline timed out after 10 seconds".into(),
214        )
215    })?
216    .map_err(|error| map_control_system_query_error(&error))?;
217    let system_identifier = parse_decimal_identity::<u64>(control_row.get(0), "system identifier")?;
218    let timeline_id = parse_decimal_identity::<u32>(control_row.get(1), "timeline ID")?;
219
220    // Keep the database, publication, and slot projection in one statement so
221    // its catalog rows come from one PostgreSQL snapshot. The JSONB rendering
222    // is deterministic and automatically includes new publication properties.
223    let row = tokio::time::timeout(
224        CONNECT_TIMEOUT,
225        client.query_opt(
226            "WITH publication_identity AS ( \
227                 SELECT p.oid::text AS publication_oid, p.pubtruncate, \
228                        jsonb_build_object( \
229                            'properties', to_jsonb(p) - ARRAY['oid', 'pubname', 'pubowner']::text[], \
230                            'tables', COALESCE( \
231                                (SELECT jsonb_agg( \
232                                     jsonb_build_array( \
233                                         c.oid::text, pt.schemaname, pt.tablename, \
234                                         pt.attnames, pt.rowfilter \
235                                     ) \
236                                     ORDER BY pt.schemaname, pt.tablename, c.oid \
237                                 ) \
238                                 FROM pg_catalog.pg_publication_tables AS pt \
239                                 LEFT JOIN pg_catalog.pg_namespace AS n \
240                                        ON n.nspname = pt.schemaname \
241                                 LEFT JOIN pg_catalog.pg_class AS c \
242                                        ON c.relnamespace = n.oid AND c.relname = pt.tablename \
243                                 WHERE pt.pubname = p.pubname), \
244                                '[]'::jsonb \
245                            ) \
246                        )::text AS definition \
247                 FROM pg_catalog.pg_publication AS p \
248                 WHERE p.pubname = $2 \
249             ) \
250             SELECT s.confirmed_flush_lsn::text, s.plugin, s.slot_type, \
251                    s.database::text, s.temporary, s.two_phase, s.failover, \
252                    s.invalidation_reason, db.oid::text, publication_identity.publication_oid, \
253                    publication_identity.definition, publication_identity.pubtruncate \
254             FROM pg_catalog.pg_replication_slots AS s \
255             CROSS JOIN pg_catalog.pg_database AS db \
256             LEFT JOIN publication_identity ON TRUE \
257             WHERE s.slot_name = $1 AND db.datname = current_database()",
258            &[&slot_name, &publication],
259        ),
260    )
261    .await
262    .map_err(|_| {
263        ConnectorError::ConnectionFailed("query replication slot timed out after 10 seconds".into())
264    })?
265    .map_err(|error| {
266        ConnectorError::ConnectionFailed(format!(
267            "query PostgreSQL replication identity: {error}"
268        ))
269    })?;
270
271    let Some(row) = row else {
272        return Ok(None);
273    };
274    let configured_plugin: Option<&str> = row.get(1);
275    let slot_type: &str = row.get(2);
276    let configured_database: Option<&str> = row.get(3);
277    let temporary: bool = row.get(4);
278    let two_phase: bool = row.get(5);
279    let failover: bool = row.get(6);
280    let invalidation_reason: Option<&str> = row.get(7);
281    validate_replication_slot(
282        slot_name,
283        plugin,
284        database,
285        configured_plugin,
286        slot_type,
287        configured_database,
288        temporary,
289        invalidation_reason,
290    )?;
291    let database_oid = parse_decimal_identity::<u32>(row.get(8), "database OID")?;
292    let publication_oid_text: Option<&str> = row.get(9);
293    let publication_oid = publication_oid_text
294        .ok_or_else(|| {
295            ConnectorError::ConfigurationError(format!(
296                "PostgreSQL publication '{publication}' does not exist"
297            ))
298        })
299        .and_then(|value| parse_decimal_identity::<u32>(value, "publication OID"))?;
300    let publication_definition: Option<&str> = row.get(10);
301    let publication_definition = publication_definition.ok_or_else(|| {
302        ConnectorError::ConfigurationError(format!(
303            "PostgreSQL publication '{publication}' has no readable definition"
304        ))
305    })?;
306    let publication_truncates: Option<bool> = row.get(11);
307    if publication_truncates != Some(false) {
308        return Err(ConnectorError::ConfigurationError(format!(
309            "PostgreSQL publication '{publication}' publishes TRUNCATE, which this CDC source cannot represent; recreate or alter it with publish='insert,update,delete'"
310        )));
311    }
312    let mut publication_digest = Sha256::new();
313    publication_digest.update(b"laminardb-postgres-publication-v1\0");
314    digest_field(&mut publication_digest, publication_definition.as_bytes());
315    tracing::info!(
316        slot = slot_name,
317        two_phase,
318        failover,
319        "using logical replication slot"
320    );
321
322    let lsn: Option<&str> = row.get(0);
323    let confirmed_flush_lsn = lsn
324        .map(|value| {
325            value.parse().map_err(|error| {
326                ConnectorError::ReadError(format!("invalid confirmed_flush_lsn: {error}"))
327            })
328        })
329        .transpose()?;
330    Ok(Some(InspectedReplicationSlot {
331        confirmed_flush_lsn,
332        binding: PostgresCheckpointBinding {
333            system_identifier,
334            timeline_id,
335            database_oid,
336            publication_oid,
337            publication_definition_sha256: format!("{:x}", publication_digest.finalize()),
338            source_config_sha256,
339            slot_plugin: configured_plugin.unwrap_or_default().to_string(),
340            slot_two_phase: two_phase,
341            slot_failover: failover,
342        },
343    }))
344}
345
346fn validate_server_version_num(version_num: u32) -> Result<(), ConnectorError> {
347    if version_num < MINIMUM_SERVER_VERSION_NUM {
348        return Err(ConnectorError::ConfigurationError(format!(
349            "PostgreSQL CDC requires PostgreSQL 17 or newer; server_version_num is {version_num}"
350        )));
351    }
352    Ok(())
353}
354
355#[cfg(not(test))]
356fn parse_decimal_identity<T>(value: &str, label: &str) -> Result<T, ConnectorError>
357where
358    T: std::str::FromStr,
359    T::Err: std::fmt::Display,
360{
361    value.parse::<T>().map_err(|error| {
362        ConnectorError::ReadError(format!("invalid PostgreSQL {label} '{value}': {error}"))
363    })
364}
365
366#[cfg(not(test))]
367fn map_control_system_query_error(error: &tokio_postgres::Error) -> ConnectorError {
368    if error.code() == Some(&tokio_postgres::error::SqlState::INSUFFICIENT_PRIVILEGE) {
369        return ConnectorError::ConfigurationError(
370            "PostgreSQL CDC must call pg_catalog.pg_control_system() and pg_catalog.pg_control_checkpoint() to bind checkpoints to a physical cluster and WAL timeline; grant the replication role pg_monitor or EXECUTE on both functions"
371                .into(),
372        );
373    }
374    ConnectorError::ConnectionFailed(format!(
375        "query PostgreSQL system identifier and timeline: {error}"
376    ))
377}
378
379fn validate_replication_slot(
380    slot_name: &str,
381    expected_plugin: &str,
382    expected_database: &str,
383    configured_plugin: Option<&str>,
384    slot_type: &str,
385    configured_database: Option<&str>,
386    temporary: bool,
387    invalidation_reason: Option<&str>,
388) -> Result<(), ConnectorError> {
389    if slot_type != "logical" || configured_plugin != Some(expected_plugin) {
390        return Err(ConnectorError::ConfigurationError(format!(
391            "PostgreSQL replication slot '{slot_name}' is not a logical {expected_plugin} slot"
392        )));
393    }
394    if configured_database != Some(expected_database) {
395        return Err(ConnectorError::ConfigurationError(format!(
396            "PostgreSQL replication slot '{slot_name}' belongs to database '{}', not configured database '{expected_database}'",
397            configured_database.unwrap_or("<none>")
398        )));
399    }
400    if temporary {
401        return Err(ConnectorError::ConfigurationError(format!(
402            "PostgreSQL replication slot '{slot_name}' is temporary and cannot provide durable recovery"
403        )));
404    }
405    if let Some(reason) = invalidation_reason {
406        return Err(ConnectorError::ReadError(format!(
407            "PostgreSQL replication slot '{slot_name}' is invalidated: {reason}"
408        )));
409    }
410    Ok(())
411}
412
413/// Builds the replication client configuration from the validated source config.
414#[must_use]
415pub(super) fn build_replication_config(
416    config: &super::config::PostgresCdcConfig,
417) -> pgwire_replication::ReplicationConfig {
418    pgwire_replication::ReplicationConfig {
419        host: config.host.clone(),
420        port: config.port,
421        user: config.username.clone(),
422        password: config.password.clone().unwrap_or_default(),
423        database: config.database.clone(),
424        tls: match config.ssl_mode {
425            crate::postgres::SslMode::Disable => pgwire_replication::TlsConfig::disabled(),
426            crate::postgres::SslMode::VerifyFull => {
427                pgwire_replication::TlsConfig::verify_full(config.ssl_ca_cert_path.clone())
428            }
429        },
430        slot: config.slot_name.clone(),
431        publication: config.publication.clone(),
432        // The exact slot/checkpoint cursor is installed by `PostgresCdcSource::start` after it
433        // validates the durable slot. A user-supplied cursor is never accepted as configuration.
434        start_lsn: pgwire_replication::Lsn::ZERO,
435        expected_recovery_identity: None,
436        stop_at_lsn: None,
437        status_interval: std::time::Duration::from_secs(1),
438        idle_wakeup_interval: std::time::Duration::from_secs(1),
439        buffer_events: 8192,
440        max_message_bytes: config.raw_wal_bytes(),
441        max_in_flight_bytes: config.raw_wal_bytes(),
442    }
443}
444
445#[cfg(test)]
446mod tests {
447    use super::{
448        build_replication_config, source_config_digest, validate_replication_slot,
449        validate_server_version_num,
450    };
451    use crate::postgres::cdc::{PostgresCdcConfig, SslMode};
452
453    #[test]
454    fn replication_config_disables_tls() {
455        let mut config = PostgresCdcConfig::default();
456        config.ssl_mode = SslMode::Disable;
457        let replication = build_replication_config(&config);
458        assert_eq!(replication.tls.mode, pgwire_replication::SslMode::Disable);
459    }
460
461    #[test]
462    fn replication_config_maps_verified_tls_and_custom_ca() {
463        let mut config = PostgresCdcConfig::default();
464        config.ssl_mode = SslMode::VerifyFull;
465        config.ssl_ca_cert_path = Some("/certs/ca.pem".into());
466
467        let replication = build_replication_config(&config);
468        assert_eq!(
469            replication.tls.mode,
470            pgwire_replication::SslMode::VerifyFull
471        );
472        assert_eq!(replication.tls.ca_pem_path, Some("/certs/ca.pem".into()));
473        assert_eq!(
474            replication.status_interval,
475            std::time::Duration::from_secs(1)
476        );
477        assert_eq!(
478            replication.idle_wakeup_interval,
479            std::time::Duration::from_secs(1)
480        );
481        assert_eq!(replication.max_message_bytes, config.raw_wal_bytes());
482        assert_eq!(replication.max_in_flight_bytes, config.raw_wal_bytes());
483    }
484
485    #[test]
486    fn replication_config_maps_connection_identity() {
487        let mut config = PostgresCdcConfig::new("pg.example.com", "mydb", "my_slot", "my_pub");
488        config.ssl_mode = SslMode::Disable;
489        config.port = 5433;
490        config.username = "replicator".to_string();
491        config.password = Some("secret".to_string());
492
493        let replication = build_replication_config(&config);
494        assert_eq!(replication.host, "pg.example.com");
495        assert_eq!(replication.port, 5433);
496        assert_eq!(replication.user, "replicator");
497        assert_eq!(replication.password, "secret");
498        assert_eq!(replication.database, "mydb");
499        assert_eq!(replication.slot, "my_slot");
500        assert_eq!(replication.publication, "my_pub");
501    }
502
503    #[test]
504    fn existing_slot_must_match_the_durable_logical_identity() {
505        validate_replication_slot(
506            "slot",
507            "pgoutput",
508            "app",
509            Some("pgoutput"),
510            "logical",
511            Some("app"),
512            false,
513            None,
514        )
515        .unwrap();
516
517        for error in [
518            validate_replication_slot(
519                "slot",
520                "pgoutput",
521                "app",
522                Some("test_decoding"),
523                "logical",
524                Some("app"),
525                false,
526                None,
527            )
528            .unwrap_err(),
529            validate_replication_slot(
530                "slot",
531                "pgoutput",
532                "app",
533                Some("pgoutput"),
534                "logical",
535                Some("other"),
536                false,
537                None,
538            )
539            .unwrap_err(),
540            validate_replication_slot(
541                "slot",
542                "pgoutput",
543                "app",
544                Some("pgoutput"),
545                "logical",
546                Some("app"),
547                true,
548                None,
549            )
550            .unwrap_err(),
551            validate_replication_slot(
552                "slot",
553                "pgoutput",
554                "app",
555                Some("pgoutput"),
556                "logical",
557                Some("app"),
558                false,
559                Some("wal_removed"),
560            )
561            .unwrap_err(),
562        ] {
563            assert!(error.to_string().contains("slot"));
564        }
565    }
566
567    #[test]
568    fn source_config_digest_is_canonical_but_semantic() {
569        let mut first = PostgresCdcConfig::default();
570        first.table_include = vec!["public.b".into(), "public.a".into(), "public.a".into()];
571        first.table_exclude = vec!["public.audit".into()];
572
573        let mut reordered = first.clone();
574        reordered.table_include = vec!["public.a".into(), "public.b".into()];
575        reordered.host = "replacement-primary".into();
576        reordered.max_buffered_bytes = 64 * 1024 * 1024;
577        assert_eq!(
578            source_config_digest(&first),
579            source_config_digest(&reordered),
580            "endpoint, capacity, order, and duplicates do not change filtering semantics"
581        );
582
583        reordered.table_exclude.push("public.private".into());
584        assert_ne!(
585            source_config_digest(&first),
586            source_config_digest(&reordered)
587        );
588    }
589
590    #[test]
591    fn server_version_is_admitted_before_pg17_slot_columns_are_used() {
592        let error = validate_server_version_num(160_012).unwrap_err();
593        assert!(error.to_string().contains("PostgreSQL 17"), "{error}");
594        validate_server_version_num(170_000).unwrap();
595        validate_server_version_num(180_001).unwrap();
596    }
597}
laminar_connectors/postgres/cdc/postgres_io.rs

laminar_connectors/postgres/cdc/
postgres_io.rs
