laminar_connectors/postgres/cdc/postgres_io/
mod.rs

1//! `PostgreSQL` logical replication connections and slot administration.
2
3mod catalog;
4pub(super) mod slots;
5
6pub(super) use catalog::{
7    in_recovery, inspect_capture_table, layout_fingerprint, layout_fingerprint_sql,
8    primary_key_index, unreadable_columns, CaptureTable, IndexKeyColumn,
9};
10
11use crate::connector::ConnectorTaskGuard;
12use crate::error::ConnectorError;
13use sha2::{Digest, Sha256};
14use slots::{SlotAttributes, SlotFacts, SlotHolder};
15
16pub(super) const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
17
18const MINIMUM_SERVER_VERSION_NUM: u32 = 170_000;
19
20/// Database-side identity that makes an engine checkpoint safe to resume. The slot's own
21/// attributes are fixed by the adoption checks rather than recorded here.
22#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
23pub(super) struct PostgresCheckpointBinding {
24    pub system_identifier: u64,
25    pub timeline_id: u32,
26    pub database_oid: u32,
27    pub publication_oid: u32,
28    pub publication_definition_sha256: String,
29    pub source_config_sha256: String,
30}
31
32/// Cluster, publication, and optional slot identity read from one catalog snapshot.
33#[derive(Debug, Clone)]
34pub(super) struct InspectedSource {
35    pub system_identifier: u64,
36    pub timeline_id: u32,
37    pub database_oid: u32,
38    pub publication_oid: u32,
39    pub publication_definition_sha256: String,
40    pub slot: Option<SlotFacts>,
41}
42
43impl InspectedSource {
44    /// The checkpoint binding of the live database and publication.
45    pub(super) fn binding(
46        &self,
47        config: &super::config::PostgresCdcConfig,
48    ) -> PostgresCheckpointBinding {
49        PostgresCheckpointBinding {
50            system_identifier: self.system_identifier,
51            timeline_id: self.timeline_id,
52            database_oid: self.database_oid,
53            publication_oid: self.publication_oid,
54            publication_definition_sha256: self.publication_definition_sha256.clone(),
55            source_config_sha256: source_config_digest(config),
56        }
57    }
58}
59
60fn digest_field(digest: &mut Sha256, value: &[u8]) {
61    digest.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes());
62    digest.update(value);
63}
64
65/// Hashes only settings that change which logical changes Laminar emits and how.
66/// Connection endpoints, snapshot mode, and buffering limits deliberately remain restartable.
67#[must_use]
68pub(super) fn source_config_digest(config: &super::config::PostgresCdcConfig) -> String {
69    let mut digest = Sha256::new();
70    digest.update(b"laminardb-postgres-cdc-source-v2\0");
71    digest_field(&mut digest, b"pgoutput");
72    digest_field(&mut digest, b"proto_version=1");
73    digest_field(&mut digest, b"messages=false");
74    digest_field(&mut digest, super::typed_rows::SESSION_OPTIONS.as_bytes());
75    digest_field(&mut digest, config.table.schema.as_bytes());
76    digest_field(&mut digest, config.table.name.as_bytes());
77    digest_field(&mut digest, config.output_mode.to_string().as_bytes());
78    format!("{:x}", digest.finalize())
79}
80
81/// Cancellation-safe control-plane connection and driver task.
82pub(super) struct ControlConnection {
83    client: tokio_postgres::Client,
84    handle: Option<tokio::task::JoinHandle<()>>,
85}
86
87impl ControlConnection {
88    #[must_use]
89    pub(super) fn client(&self) -> &tokio_postgres::Client {
90        &self.client
91    }
92
93    pub(super) async fn close(mut self) {
94        if let Some(handle) = self.handle.take() {
95            handle.abort();
96            let _ = handle.await;
97        }
98    }
99}
100
101impl Drop for ControlConnection {
102    fn drop(&mut self) {
103        if let Some(handle) = self.handle.take() {
104            handle.abort();
105        }
106    }
107}
108
109/// Opens a regular connection with the canonical CDC session settings.
110///
111/// # Errors
112///
113/// Returns an error when TLS configuration is invalid or the connection cannot be opened.
114pub(super) async fn connect(
115    config: &super::config::PostgresCdcConfig,
116    application_name: &str,
117    driver_guard: ConnectorTaskGuard,
118) -> Result<ControlConnection, ConnectorError> {
119    use crate::postgres::SslMode;
120
121    let pg_config = config.control_connection_config(application_name)?;
122    match config.ssl_mode {
123        SslMode::Disable => {
124            let (client, connection) =
125                tokio::time::timeout(CONNECT_TIMEOUT, pg_config.connect(tokio_postgres::NoTls))
126                    .await
127                    .map_err(|_| {
128                        ConnectorError::ConnectionFailed(
129                            "PostgreSQL connect timed out after 10 seconds".into(),
130                        )
131                    })?
132                    .map_err(|error| {
133                        ConnectorError::ConnectionFailed(format!("PostgreSQL connect: {error}"))
134                    })?;
135            let handle = tokio::spawn(async move {
136                let _driver_guard = driver_guard;
137                if let Err(error) = connection.await {
138                    tracing::error!(%error, "PostgreSQL control-plane connection error");
139                }
140            });
141            Ok(ControlConnection {
142                client,
143                handle: Some(handle),
144            })
145        }
146        SslMode::VerifyFull => {
147            let tls = crate::postgres::make_rustls_connector(config.ssl_ca_cert_path.as_deref())?;
148            let (client, connection) =
149                tokio::time::timeout(CONNECT_TIMEOUT, pg_config.connect(tls))
150                    .await
151                    .map_err(|_| {
152                        ConnectorError::ConnectionFailed(
153                            "PostgreSQL TLS connect timed out after 10 seconds".into(),
154                        )
155                    })?
156                    .map_err(|error| {
157                        ConnectorError::ConnectionFailed(format!("PostgreSQL TLS connect: {error}"))
158                    })?;
159            let handle = tokio::spawn(async move {
160                let _driver_guard = driver_guard;
161                if let Err(error) = connection.await {
162                    tracing::error!(%error, "PostgreSQL control-plane TLS connection error");
163                }
164            });
165            Ok(ControlConnection {
166                client,
167                handle: Some(handle),
168            })
169        }
170    }
171}
172
173/// Inspects the cluster, the publication, and the slot named `slot` without mutating any of them.
174///
175/// A present slot carries its holder and the adoption checks' verdict.
176///
177/// # Errors
178///
179/// Returns an error when identity validation, publication admission, or LSN parsing fails.
180pub(super) async fn inspect_source(
181    client: &tokio_postgres::Client,
182    config: &super::config::PostgresCdcConfig,
183    slot: Option<&str>,
184) -> Result<InspectedSource, ConnectorError> {
185    let (system_identifier, timeline_id) = read_system_identity(client).await?;
186
187    // Keep the database, publication, and slot projection in one statement so its catalog rows
188    // come from one PostgreSQL snapshot. The JSONB rendering is deterministic and automatically
189    // includes new publication properties.
190    let row = tokio::time::timeout(
191        CONNECT_TIMEOUT,
192        client.query_one(
193            "WITH publication_identity AS ( \
194                 SELECT p.oid::text AS publication_oid, p.pubtruncate, \
195                        jsonb_build_object( \
196                            'properties', to_jsonb(p) - ARRAY['oid', 'pubname', 'pubowner']::text[], \
197                            'tables', COALESCE( \
198                                (SELECT jsonb_agg( \
199                                     jsonb_build_array( \
200                                         c.oid::text, pt.schemaname, pt.tablename, \
201                                         pt.attnames, pt.rowfilter \
202                                     ) \
203                                     ORDER BY pt.schemaname, pt.tablename, c.oid \
204                                 ) \
205                                 FROM pg_catalog.pg_publication_tables AS pt \
206                                 LEFT JOIN pg_catalog.pg_namespace AS n \
207                                        ON n.nspname = pt.schemaname \
208                                 LEFT JOIN pg_catalog.pg_class AS c \
209                                        ON c.relnamespace = n.oid AND c.relname = pt.tablename \
210                                 WHERE pt.pubname = p.pubname), \
211                                '[]'::jsonb \
212                            ) \
213                        )::text AS definition \
214                 FROM pg_catalog.pg_publication AS p \
215                 WHERE p.pubname = $2 \
216             ) \
217             SELECT db.oid::text, publication_identity.publication_oid, \
218                    publication_identity.definition, publication_identity.pubtruncate, \
219                    s.slot_name IS NOT NULL, s.confirmed_flush_lsn::text, s.slot_type, s.plugin, \
220                    s.database::text, s.datoid::text, s.temporary, s.two_phase, s.failover, \
221                    s.synced, s.invalidation_reason, s.conflicting, s.wal_status, \
222                    s.active_pid, a.application_name, a.client_addr::text \
223             FROM pg_catalog.pg_database AS db \
224             LEFT JOIN pg_catalog.pg_replication_slots AS s ON s.slot_name = $1 \
225             LEFT JOIN pg_catalog.pg_stat_activity AS a ON a.pid = s.active_pid \
226             LEFT JOIN publication_identity ON TRUE \
227             WHERE db.datname = current_database()",
228            &[&slot, &config.publication],
229        ),
230    )
231    .await
232    .map_err(|_| {
233        ConnectorError::ConnectionFailed("query replication slot timed out after 10 seconds".into())
234    })?
235    .map_err(|error| {
236        ConnectorError::ConnectionFailed(format!(
237            "query PostgreSQL replication identity: {error}"
238        ))
239    })?;
240
241    let (database_oid, publication_oid, publication_definition_sha256) =
242        read_publication_identity(&row, &config.publication)?;
243    let slot = match slot {
244        Some(name) if row.get::<_, bool>(4) => {
245            Some(read_slot(&row, name, &config.database, database_oid)?)
246        }
247        _ => None,
248    };
249    Ok(InspectedSource {
250        system_identifier,
251        timeline_id,
252        database_oid,
253        publication_oid,
254        publication_definition_sha256,
255        slot,
256    })
257}
258
259/// The slot columns of [`inspect_source`]'s row, from column 5 on.
260fn read_slot(
261    row: &tokio_postgres::Row,
262    name: &str,
263    database: &str,
264    database_oid: u32,
265) -> Result<SlotFacts, ConnectorError> {
266    let confirmed: Option<&str> = row.get(5);
267    let confirmed_flush_lsn = confirmed
268        .map(|value| {
269            value.parse().map_err(|error| {
270                ConnectorError::ReadError(format!("invalid confirmed_flush_lsn: {error}"))
271            })
272        })
273        .transpose()?;
274    let datoid: Option<&str> = row.get(9);
275    let attributes = SlotAttributes {
276        slot_type: row.get(6),
277        plugin: row.get(7),
278        database: row.get(8),
279        database_oid: datoid
280            .map(|oid| parse_decimal_identity::<u32>(oid, "slot database OID"))
281            .transpose()?,
282        temporary: row.get::<_, Option<bool>>(10).unwrap_or(false),
283        two_phase: row.get::<_, Option<bool>>(11).unwrap_or(false),
284        failover: row.get::<_, Option<bool>>(12).unwrap_or(false),
285        synced: row.get::<_, Option<bool>>(13).unwrap_or(false),
286        invalidation_reason: row.get(14),
287        conflicting: row.get::<_, Option<bool>>(15).unwrap_or(false),
288        wal_status: row.get(16),
289    };
290    let holder = row.get::<_, Option<i32>>(17).map(|pid| SlotHolder {
291        pid,
292        application_name: row.get::<_, Option<String>>(18).unwrap_or_default(),
293        client_addr: row.get(19),
294    });
295    if attributes.wal_status.as_deref() == Some("unreserved") {
296        tracing::warn!(
297            slot = name,
298            "PostgreSQL replication slot no longer reserves all the WAL it needs; the next \
299             checkpoint removes some of it unless the slot catches up (wal_status = unreserved, \
300             max_slot_wal_keep_size)"
301        );
302    }
303    Ok(SlotFacts {
304        confirmed_flush_lsn,
305        holder,
306        unusable: attributes.problem(database, database_oid),
307    })
308}
309
310fn read_publication_identity(
311    row: &tokio_postgres::Row,
312    publication: &str,
313) -> Result<(u32, u32, String), ConnectorError> {
314    let database_oid = parse_decimal_identity::<u32>(row.get(0), "database OID")?;
315    let publication_oid_text: Option<&str> = row.get(1);
316    let publication_oid = publication_oid_text
317        .ok_or_else(|| {
318            ConnectorError::ConfigurationError(format!(
319                "PostgreSQL publication '{publication}' does not exist"
320            ))
321        })
322        .and_then(|value| parse_decimal_identity::<u32>(value, "publication OID"))?;
323    let publication_definition: Option<&str> = row.get(2);
324    let publication_definition = publication_definition.ok_or_else(|| {
325        ConnectorError::ConfigurationError(format!(
326            "PostgreSQL publication '{publication}' has no readable definition"
327        ))
328    })?;
329    let publication_truncates: Option<bool> = row.get(3);
330    if publication_truncates != Some(true) {
331        return Err(ConnectorError::ConfigurationError(format!(
332            "PostgreSQL publication '{publication}' must publish TRUNCATE so a source truncation \
333             stops the pipeline instead of silently diverging the target; use the default \
334             publish='insert, update, delete, truncate'"
335        )));
336    }
337    let mut publication_digest = Sha256::new();
338    publication_digest.update(b"laminardb-postgres-publication-v1\0");
339    digest_field(&mut publication_digest, publication_definition.as_bytes());
340    Ok((
341        database_oid,
342        publication_oid,
343        format!("{:x}", publication_digest.finalize()),
344    ))
345}
346
347async fn read_system_identity(
348    client: &tokio_postgres::Client,
349) -> Result<(u64, u32), ConnectorError> {
350    let version_row = tokio::time::timeout(
351        CONNECT_TIMEOUT,
352        client.query_one("SELECT current_setting('server_version_num')", &[]),
353    )
354    .await
355    .map_err(|_| {
356        ConnectorError::ConnectionFailed(
357            "query PostgreSQL server version timed out after 10 seconds".into(),
358        )
359    })?
360    .map_err(|error| {
361        ConnectorError::ConnectionFailed(format!("query PostgreSQL server version: {error}"))
362    })?;
363    let version_text: &str = version_row.try_get(0).map_err(|error| {
364        ConnectorError::ReadError(format!("read PostgreSQL server version: {error}"))
365    })?;
366    let version_num = version_text.parse::<u32>().map_err(|error| {
367        ConnectorError::ReadError(format!(
368            "invalid PostgreSQL server_version_num '{version_text}': {error}"
369        ))
370    })?;
371    validate_server_version_num(version_num)?;
372
373    let control_row = tokio::time::timeout(
374        CONNECT_TIMEOUT,
375        client.query_one(
376            "SELECT control_system.system_identifier::text, control_checkpoint.timeline_id::text \
377             FROM pg_catalog.pg_control_system() AS control_system \
378             CROSS JOIN pg_catalog.pg_control_checkpoint() AS control_checkpoint",
379            &[],
380        ),
381    )
382    .await
383    .map_err(|_| {
384        ConnectorError::ConnectionFailed(
385            "query PostgreSQL system identifier and timeline timed out after 10 seconds".into(),
386        )
387    })?
388    .map_err(|error| map_control_system_query_error(&error))?;
389    let system_identifier = parse_decimal_identity::<u64>(
390        control_row.try_get(0).map_err(|error| {
391            ConnectorError::ReadError(format!("read PostgreSQL system identifier: {error}"))
392        })?,
393        "system identifier",
394    )?;
395    let timeline_id = parse_decimal_identity::<u32>(
396        control_row.try_get(1).map_err(|error| {
397            ConnectorError::ReadError(format!("read PostgreSQL timeline: {error}"))
398        })?,
399        "timeline ID",
400    )?;
401
402    Ok((system_identifier, timeline_id))
403}
404
405/// WAL retention state of a slot (`pg_replication_slots.wal_status`, `safe_wal_size`).
406///
407/// # Errors
408///
409/// Returns an error when the slot is missing or the query fails.
410pub(super) async fn slot_wal_status(
411    client: &tokio_postgres::Client,
412    slot_name: &str,
413) -> Result<(String, Option<i64>), ConnectorError> {
414    let row = tokio::time::timeout(
415        CONNECT_TIMEOUT,
416        client.query_opt(
417            "SELECT wal_status, safe_wal_size FROM pg_catalog.pg_replication_slots \
418             WHERE slot_name = $1",
419            &[&slot_name],
420        ),
421    )
422    .await
423    .map_err(|_| {
424        ConnectorError::ConnectionFailed("query slot WAL status timed out after 10 seconds".into())
425    })?
426    .map_err(|error| ConnectorError::ConnectionFailed(format!("query slot WAL status: {error}")))?
427    .ok_or_else(|| {
428        ConnectorError::ReadError(format!(
429            "PostgreSQL replication slot '{slot_name}' disappeared"
430        ))
431    })?;
432    Ok((
433        row.get::<_, Option<String>>(0).unwrap_or_default(),
434        row.get(1),
435    ))
436}
437
438fn validate_server_version_num(version_num: u32) -> Result<(), ConnectorError> {
439    if version_num < MINIMUM_SERVER_VERSION_NUM {
440        return Err(ConnectorError::ConfigurationError(format!(
441            "PostgreSQL CDC requires PostgreSQL 17 or newer; server_version_num is {version_num}"
442        )));
443    }
444    Ok(())
445}
446
447/// Whether a failed statement may succeed on a new connection: no SQLSTATE (the connection was
448/// lost), a connection exception (`08`), or a server shutdown or restart (`57P0`).
449pub(super) fn is_connection_failure(sqlstate: Option<&str>) -> bool {
450    sqlstate.is_none_or(|code| code.starts_with("08") || code.starts_with("57P0"))
451}
452
453fn parse_decimal_identity<T>(value: &str, label: &str) -> Result<T, ConnectorError>
454where
455    T: std::str::FromStr,
456    T::Err: std::fmt::Display,
457{
458    value.parse::<T>().map_err(|error| {
459        ConnectorError::ReadError(format!("invalid PostgreSQL {label} '{value}': {error}"))
460    })
461}
462
463fn map_control_system_query_error(error: &tokio_postgres::Error) -> ConnectorError {
464    if error.code() == Some(&tokio_postgres::error::SqlState::INSUFFICIENT_PRIVILEGE) {
465        return ConnectorError::ConfigurationError(
466            "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"
467                .into(),
468        );
469    }
470    ConnectorError::ConnectionFailed(format!(
471        "query PostgreSQL system identifier and timeline: {error}"
472    ))
473}
474
475/// Builds the replication client configuration for `slot` from the validated source config.
476#[must_use]
477pub(super) fn build_replication_config(
478    config: &super::config::PostgresCdcConfig,
479    slot: &str,
480    application_name: &str,
481) -> pgwire_replication::ReplicationConfig {
482    pgwire_replication::ReplicationConfig {
483        host: config.host.clone(),
484        port: config.port,
485        user: config.username.clone(),
486        password: config.password.clone().unwrap_or_default(),
487        database: config.database.clone(),
488        tls: match config.ssl_mode {
489            crate::postgres::SslMode::Disable => pgwire_replication::TlsConfig::disabled(),
490            crate::postgres::SslMode::VerifyFull => {
491                pgwire_replication::TlsConfig::verify_full(config.ssl_ca_cert_path.clone())
492            }
493        },
494        slot: slot.to_string(),
495        publication: config.publication.clone(),
496        // The reader launch sets the position once the source has adopted or created its slot;
497        // a user-supplied cursor is never accepted as configuration.
498        start_lsn: pgwire_replication::Lsn::ZERO,
499        expected_recovery_identity: None,
500        stop_at_lsn: None,
501        status_interval: std::time::Duration::from_secs(1),
502        idle_wakeup_interval: std::time::Duration::from_secs(1),
503        buffer_events: 8192,
504        max_message_bytes: config.raw_wal_bytes(),
505        session_options: Some(super::typed_rows::SESSION_OPTIONS.into()),
506        application_name: application_name.to_string(),
507        max_in_flight_bytes: config.raw_wal_bytes(),
508    }
509}
510
511#[cfg(test)]
512mod tests;
laminar_connectors/postgres/cdc/postgres_io/mod.rs

laminar_connectors/postgres/cdc/postgres_io/
mod.rs
