laminar_connectors/postgres/cdc/
config.rs

1//! `PostgreSQL` CDC source connector configuration.
2//!
3//! Provides [`PostgresCdcConfig`] with all settings needed to connect to
4//! a `PostgreSQL` database and stream logical replication changes.
5
6use std::path::PathBuf;
7
8use crate::config::ConnectorConfig;
9use crate::error::ConnectorError;
10use crate::postgres::SslMode;
11
12const REMOVED_CONFIG_KEYS: &[&str] = &[
13    "backpressure.high.watermark",
14    "keepalive.interval.ms",
15    "max.buffered.events",
16    "max.poll.records",
17    "poll.timeout.ms",
18    "snapshot.mode",
19    "start.lsn",
20    "wal.sender.timeout.ms",
21];
22
23const DEFAULT_BUFFERED_BYTES: usize = 256 * 1024 * 1024;
24const MIN_BUFFERED_BYTES: usize = 1024 * 1024;
25const MAX_BUFFERED_BYTES: usize = 4 * 1024 * 1024 * 1024;
26const WORKING_SET_STAGES: usize = 3;
27
28const ALLOWED_CONFIG_KEYS: &[&str] = &[
29    "_arrow_schema",
30    "database",
31    "host",
32    "laminar.source.name",
33    "max.buffered.bytes",
34    "password",
35    "port",
36    "publication",
37    "slot.name",
38    "ssl.ca.cert.path",
39    "ssl.mode",
40    "table.exclude",
41    "table.include",
42    "username",
43];
44
45/// Configuration for the `PostgreSQL` CDC source connector.
46#[derive(Debug, Clone)]
47pub struct PostgresCdcConfig {
48    // ── Connection ──
49    /// `PostgreSQL` host address.
50    pub host: String,
51
52    /// `PostgreSQL` port.
53    pub port: u16,
54
55    /// Database name.
56    pub database: String,
57
58    /// Username for authentication.
59    pub username: String,
60
61    /// Password for authentication.
62    pub password: Option<String>,
63
64    /// SSL mode for the connection.
65    pub ssl_mode: SslMode,
66
67    /// Optional PEM file containing trusted CA certificates.
68    pub ssl_ca_cert_path: Option<PathBuf>,
69
70    // ── Replication ──
71    /// Name of the logical replication slot.
72    pub slot_name: String,
73
74    /// Name of the publication to subscribe to.
75    pub publication: String,
76
77    // ── Schema ──
78    /// Tables to include (empty = all tables in publication).
79    pub table_include: Vec<String>,
80
81    /// Tables to exclude from replication.
82    pub table_exclude: Vec<String>,
83
84    /// Total connector-owned payload budget across raw WAL, decoded state, and Arrow construction.
85    pub max_buffered_bytes: usize,
86}
87
88impl Default for PostgresCdcConfig {
89    fn default() -> Self {
90        Self {
91            host: "localhost".to_string(),
92            port: 5432,
93            database: "postgres".to_string(),
94            username: "postgres".to_string(),
95            password: None,
96            ssl_mode: SslMode::VerifyFull,
97            ssl_ca_cert_path: None,
98            slot_name: "laminar_slot".to_string(),
99            publication: "laminar_pub".to_string(),
100            table_include: Vec::new(),
101            table_exclude: Vec::new(),
102            max_buffered_bytes: DEFAULT_BUFFERED_BYTES,
103        }
104    }
105}
106
107impl PostgresCdcConfig {
108    /// Decoded-stage high watermark used to stop admitting raw WAL before the hard limit.
109    #[must_use]
110    pub(super) fn decoded_high_watermark_bytes(&self) -> usize {
111        self.decoded_event_bytes()
112            .saturating_sub(self.decoded_event_bytes() / 5)
113    }
114
115    /// Raw pgwire/frame ownership share of the private working-set limit.
116    #[must_use]
117    pub(crate) fn raw_wal_bytes(&self) -> usize {
118        self.max_buffered_bytes / WORKING_SET_STAGES
119    }
120
121    /// Decoded transaction ownership share of the private working-set limit.
122    #[must_use]
123    pub(crate) fn decoded_event_bytes(&self) -> usize {
124        self.max_buffered_bytes / WORKING_SET_STAGES
125    }
126
127    /// Arrow construction share, including division remainder.
128    #[must_use]
129    pub(crate) fn arrow_build_bytes(&self) -> usize {
130        self.max_buffered_bytes
131            .saturating_sub(self.raw_wal_bytes())
132            .saturating_sub(self.decoded_event_bytes())
133    }
134
135    pub(crate) fn normalize_table_filters(&mut self) {
136        normalize_table_list(&mut self.table_include);
137        normalize_table_list(&mut self.table_exclude);
138    }
139
140    /// Creates a new config with required fields.
141    #[must_use]
142    pub fn new(host: &str, database: &str, slot_name: &str, publication: &str) -> Self {
143        Self {
144            host: host.to_string(),
145            database: database.to_string(),
146            slot_name: slot_name.to_string(),
147            publication: publication.to_string(),
148            ..Self::default()
149        }
150    }
151
152    /// Builds the typed control-plane connection configuration.
153    ///
154    /// Using setters keeps credentials and database names out of libpq-style
155    /// tokenization, so whitespace, quotes, and backslashes are data rather
156    /// than connection-string syntax.
157    pub(super) fn control_connection_config(
158        &self,
159    ) -> Result<tokio_postgres::Config, ConnectorError> {
160        self.validate()?;
161
162        let mut config = tokio_postgres::Config::new();
163        config
164            .host(&self.host)
165            .port(self.port)
166            .dbname(&self.database)
167            .user(&self.username)
168            .ssl_mode(match self.ssl_mode {
169                SslMode::Disable => tokio_postgres::config::SslMode::Disable,
170                SslMode::VerifyFull => tokio_postgres::config::SslMode::Require,
171            })
172            .connect_timeout(super::postgres_io::CONNECT_TIMEOUT);
173        if let Some(password) = &self.password {
174            config.password(password);
175        }
176        Ok(config)
177    }
178
179    /// Parses configuration from a generic [`ConnectorConfig`].
180    ///
181    /// # Errors
182    ///
183    /// Returns `ConnectorError` if required keys are missing or values are
184    /// invalid.
185    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
186        Self::reject_removed_keys(config)?;
187        config.reject_unknown_properties(ALLOWED_CONFIG_KEYS, "PostgreSQL CDC")?;
188
189        let mut cfg = Self {
190            host: config.require("host")?.to_string(),
191            database: config.require("database")?.to_string(),
192            slot_name: config.require("slot.name")?.to_string(),
193            publication: config.require("publication")?.to_string(),
194            ssl_mode: config
195                .get_parsed::<SslMode>("ssl.mode")?
196                .unwrap_or_default(),
197            ..Self::default()
198        };
199
200        if let Some(port) = config.get("port") {
201            cfg.port = crate::config::parse_port(port)?;
202        }
203        if let Some(user) = config.get("username") {
204            cfg.username = user.to_string();
205        }
206        cfg.password = config.get("password").map(String::from);
207        cfg.ssl_ca_cert_path = config.get("ssl.ca.cert.path").map(PathBuf::from);
208
209        if let Some(tables) = config.get("table.include") {
210            cfg.table_include = tables.split(',').map(str::to_string).collect();
211        }
212        if let Some(tables) = config.get("table.exclude") {
213            cfg.table_exclude = tables.split(',').map(str::to_string).collect();
214        }
215        if let Some(max) = config.get_parsed::<usize>("max.buffered.bytes")? {
216            cfg.max_buffered_bytes = max;
217        }
218        cfg.normalize_table_filters();
219        cfg.validate()?;
220        Ok(cfg)
221    }
222
223    fn reject_removed_keys(config: &ConnectorConfig) -> Result<(), ConnectorError> {
224        if let Some(key) = REMOVED_CONFIG_KEYS
225            .iter()
226            .find(|key| config.get(key).is_some())
227        {
228            let reason = match *key {
229                "start.lsn" => {
230                    "recovery cursors are owned by a validated engine checkpoint or the durable slot"
231                }
232                "max.buffered.events" => {
233                    "resource ownership is bounded by max.buffered.bytes instead"
234                }
235                _ => "the connector did not execute it",
236            };
237            return Err(ConnectorError::ConfigurationError(format!(
238                "PostgreSQL CDC property '{key}' is not supported: {reason}"
239            )));
240        }
241        Ok(())
242    }
243
244    /// Validates the configuration.
245    ///
246    /// # Errors
247    ///
248    /// Returns `ConnectorError::ConfigurationError` for invalid settings.
249    pub fn validate(&self) -> Result<(), ConnectorError> {
250        crate::config::require_non_empty(&self.host, "host")?;
251        crate::config::require_non_empty(&self.database, "database")?;
252        crate::config::require_non_empty(&self.username, "username")?;
253        crate::config::require_non_empty(&self.slot_name, "slot.name")?;
254        crate::config::require_non_empty(&self.publication, "publication")?;
255        for (value, label) in [
256            (&self.host, "host"),
257            (&self.database, "database"),
258            (&self.username, "username"),
259            (&self.slot_name, "slot.name"),
260            (&self.publication, "publication"),
261        ] {
262            if value.contains('\0') {
263                return Err(ConnectorError::ConfigurationError(format!(
264                    "{label} must not contain NUL"
265                )));
266            }
267        }
268        if self.slot_name.len() > 63
269            || !self
270                .slot_name
271                .bytes()
272                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
273        {
274            return Err(ConnectorError::ConfigurationError(
275                "slot.name must be at most 63 bytes and contain only lower-case ASCII letters, digits, or underscore"
276                    .into(),
277            ));
278        }
279        if self.publication.len() > 63
280            || !self
281                .publication
282                .bytes()
283                .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
284            || self.publication.as_bytes()[0].is_ascii_digit()
285        {
286            return Err(ConnectorError::ConfigurationError(
287                "publication must start with a lower-case ASCII letter or underscore and contain only lower-case ASCII letters, digits, or underscore (63 bytes maximum)"
288                    .into(),
289            ));
290        }
291        if self
292            .password
293            .as_deref()
294            .is_some_and(|value| value.contains('\0'))
295        {
296            return Err(ConnectorError::ConfigurationError(
297                "password must not contain NUL".into(),
298            ));
299        }
300        if self.port == 0 {
301            return Err(ConnectorError::ConfigurationError(
302                "port must be > 0".to_string(),
303            ));
304        }
305        if !(MIN_BUFFERED_BYTES..=MAX_BUFFERED_BYTES).contains(&self.max_buffered_bytes) {
306            return Err(ConnectorError::ConfigurationError(format!(
307                "max.buffered.bytes must be between {MIN_BUFFERED_BYTES} and {MAX_BUFFERED_BYTES}"
308            )));
309        }
310        if self
311            .ssl_ca_cert_path
312            .as_ref()
313            .is_some_and(|path| path.as_os_str().is_empty())
314        {
315            return Err(ConnectorError::ConfigurationError(
316                "ssl.ca.cert.path must not be empty".to_string(),
317            ));
318        }
319        if self.ssl_mode == SslMode::Disable && self.ssl_ca_cert_path.is_some() {
320            return Err(ConnectorError::ConfigurationError(
321                "ssl.ca.cert.path requires ssl.mode=verify-full".to_string(),
322            ));
323        }
324        for (label, tables) in [
325            ("table.include", &self.table_include),
326            ("table.exclude", &self.table_exclude),
327        ] {
328            for table in tables {
329                let Some((schema, relation)) = table.split_once('.') else {
330                    return Err(ConnectorError::ConfigurationError(format!(
331                        "{label} entry '{table}' must be schema-qualified as schema.table"
332                    )));
333                };
334                if table.trim() != table || schema.is_empty() || relation.is_empty() {
335                    return Err(ConnectorError::ConfigurationError(format!(
336                        "{label} entry '{table}' must contain nonempty schema and table names"
337                    )));
338                }
339            }
340        }
341        Ok(())
342    }
343
344    /// Returns whether a table should be included based on include/exclude lists.
345    #[must_use]
346    pub(crate) fn should_include_table(&self, table: &str) -> bool {
347        debug_assert!(self.table_include.is_sorted());
348        debug_assert!(self.table_exclude.is_sorted());
349        if self
350            .table_exclude
351            .binary_search_by(|candidate| candidate.as_str().cmp(table))
352            .is_ok()
353        {
354            return false;
355        }
356        if self.table_include.is_empty() {
357            return true;
358        }
359        self.table_include
360            .binary_search_by(|candidate| candidate.as_str().cmp(table))
361            .is_ok()
362    }
363}
364
365fn normalize_table_list(tables: &mut Vec<String>) {
366    for table in tables.iter_mut() {
367        *table = table.trim().to_string();
368    }
369    tables.retain(|table| !table.is_empty());
370    tables.sort_unstable();
371    tables.dedup();
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377
378    fn connector_config() -> ConnectorConfig {
379        let mut config = ConnectorConfig::new("postgres-cdc");
380        config.set("host", "localhost");
381        config.set("database", "db");
382        config.set("slot.name", "s");
383        config.set("publication", "p");
384        config.set("ssl.mode", "disable");
385        config
386    }
387
388    #[test]
389    fn test_default_config() {
390        let cfg = PostgresCdcConfig::default();
391        assert_eq!(cfg.host, "localhost");
392        assert_eq!(cfg.port, 5432);
393        assert_eq!(cfg.database, "postgres");
394        assert_eq!(cfg.slot_name, "laminar_slot");
395        assert_eq!(cfg.publication, "laminar_pub");
396        assert_eq!(cfg.ssl_mode, SslMode::VerifyFull);
397        assert!(cfg.validate().is_ok());
398    }
399
400    #[test]
401    fn replication_identity_rejects_invalid_slot_and_nul() {
402        let mut cfg = PostgresCdcConfig::default();
403        cfg.slot_name = "Mixed-Case".into();
404        assert!(cfg
405            .validate()
406            .unwrap_err()
407            .to_string()
408            .contains("slot.name"));
409
410        cfg.slot_name = "valid_slot".into();
411        cfg.publication = "bad\0publication".into();
412        assert!(cfg.validate().unwrap_err().to_string().contains("NUL"));
413
414        cfg.publication = "Mixed-Publication".into();
415        assert!(cfg
416            .validate()
417            .unwrap_err()
418            .to_string()
419            .contains("publication"));
420    }
421
422    #[test]
423    fn test_new_config() {
424        let cfg = PostgresCdcConfig::new("db.example.com", "mydb", "my_slot", "my_pub");
425        assert_eq!(cfg.host, "db.example.com");
426        assert_eq!(cfg.database, "mydb");
427        assert_eq!(cfg.slot_name, "my_slot");
428        assert_eq!(cfg.publication, "my_pub");
429        assert_eq!(cfg.ssl_mode, SslMode::VerifyFull);
430    }
431
432    #[test]
433    fn typed_control_config_preserves_adversarial_values() {
434        let mut cfg =
435            PostgresCdcConfig::new(" db\\host' ", " db name'\\ ", "valid_slot", "publication");
436        cfg.username = " user name'\\ ".into();
437        cfg.password = Some(" password with 'quotes' and \\slashes\\ ".into());
438
439        let control = cfg.control_connection_config().unwrap();
440        assert_eq!(
441            control.get_hosts(),
442            &[tokio_postgres::config::Host::Tcp(" db\\host' ".into())]
443        );
444        assert_eq!(control.get_ports(), &[5432]);
445        assert_eq!(control.get_dbname(), Some(" db name'\\ "));
446        assert_eq!(control.get_user(), Some(" user name'\\ "));
447        assert_eq!(
448            control.get_password(),
449            Some(" password with 'quotes' and \\slashes\\ ".as_bytes())
450        );
451        assert_eq!(
452            control.get_ssl_mode(),
453            tokio_postgres::config::SslMode::Require
454        );
455        assert_eq!(
456            control.get_connect_timeout().copied(),
457            Some(crate::postgres::cdc::postgres_io::CONNECT_TIMEOUT)
458        );
459    }
460
461    #[test]
462    fn typed_control_config_maps_disabled_tls_exactly() {
463        let mut cfg = PostgresCdcConfig::default();
464        cfg.ssl_mode = SslMode::Disable;
465        assert_eq!(
466            cfg.control_connection_config().unwrap().get_ssl_mode(),
467            tokio_postgres::config::SslMode::Disable
468        );
469    }
470
471    #[test]
472    fn test_from_connector_config() {
473        let mut config = ConnectorConfig::new("postgres-cdc");
474        config.set("host", "pg.local");
475        config.set("database", "testdb");
476        config.set("slot.name", "test_slot");
477        config.set("publication", "test_pub");
478        config.set("ssl.mode", "disable");
479        config.set("port", "5433");
480        config.set("max.buffered.bytes", "67108864");
481
482        let cfg = PostgresCdcConfig::from_config(&config).unwrap();
483        assert_eq!(cfg.host, "pg.local");
484        assert_eq!(cfg.port, 5433);
485        assert_eq!(cfg.database, "testdb");
486        assert_eq!(cfg.max_buffered_bytes, 64 * 1024 * 1024);
487    }
488
489    #[test]
490    fn total_byte_budget_is_partitioned_without_loss() {
491        let mut cfg = PostgresCdcConfig::default();
492        cfg.max_buffered_bytes = MIN_BUFFERED_BYTES;
493        assert_eq!(
494            cfg.decoded_high_watermark_bytes(),
495            cfg.decoded_event_bytes() - cfg.decoded_event_bytes() / 5
496        );
497        assert_eq!(
498            cfg.raw_wal_bytes() + cfg.decoded_event_bytes() + cfg.arrow_build_bytes(),
499            MIN_BUFFERED_BYTES
500        );
501    }
502
503    #[test]
504    fn total_byte_budget_rejects_values_outside_the_operational_range() {
505        for bytes in [MIN_BUFFERED_BYTES - 1, MAX_BUFFERED_BYTES + 1] {
506            let mut cfg = PostgresCdcConfig::default();
507            cfg.max_buffered_bytes = bytes;
508            assert!(cfg.validate().is_err(), "{bytes}");
509        }
510    }
511
512    #[test]
513    fn test_from_config_missing_required() {
514        let config = ConnectorConfig::new("postgres-cdc");
515        assert!(PostgresCdcConfig::from_config(&config).is_err());
516    }
517
518    #[test]
519    fn test_from_config_invalid_port() {
520        let mut config = connector_config();
521        config.set("port", "not_a_number");
522        assert!(PostgresCdcConfig::from_config(&config).is_err());
523    }
524
525    #[test]
526    fn omitted_ssl_mode_uses_verified_tls() {
527        let mut config = ConnectorConfig::new("postgres-cdc");
528        config.set("host", "localhost");
529        config.set("database", "db");
530        config.set("slot.name", "s");
531        config.set("publication", "p");
532        let config = PostgresCdcConfig::from_config(&config).unwrap();
533        assert_eq!(config.ssl_mode, SslMode::VerifyFull);
534    }
535
536    #[test]
537    fn unknown_properties_are_rejected_deterministically() {
538        let mut config = connector_config();
539        config.set("z.invalid", "1");
540        config.set("a.invalid", "2");
541        let error = PostgresCdcConfig::from_config(&config).unwrap_err();
542        assert!(error.to_string().contains("a.invalid"), "{error}");
543    }
544
545    #[test]
546    fn engine_metadata_properties_are_admitted() {
547        let mut config = connector_config();
548        config.set("laminar.source.name", "orders");
549        config.set("_arrow_schema", "engine-owned");
550        PostgresCdcConfig::from_config(&config).unwrap();
551    }
552
553    #[test]
554    fn test_validate_empty_host() {
555        let mut cfg = PostgresCdcConfig::default();
556        cfg.host = String::new();
557        assert!(cfg.validate().is_err());
558    }
559
560    #[test]
561    fn removed_properties_are_rejected_explicitly() {
562        for key in REMOVED_CONFIG_KEYS {
563            let mut config = ConnectorConfig::new("postgres-cdc");
564            config.set("host", "localhost");
565            config.set("database", "db");
566            config.set("slot.name", "s");
567            config.set("publication", "p");
568            config.set(*key, "removed-value");
569            let error = PostgresCdcConfig::from_config(&config).unwrap_err();
570            assert!(error.to_string().contains(key));
571        }
572    }
573
574    #[test]
575    fn test_table_filtering() {
576        let mut cfg = PostgresCdcConfig::default();
577        // No filters → include all
578        assert!(cfg.should_include_table("public.users"));
579
580        // Include list
581        cfg.table_include = vec!["public.users".to_string(), "public.orders".to_string()];
582        cfg.normalize_table_filters();
583        assert!(cfg.should_include_table("public.users"));
584        assert!(!cfg.should_include_table("public.logs"));
585
586        // Exclude overrides include
587        cfg.table_exclude = vec!["public.users".to_string()];
588        assert!(!cfg.should_include_table("public.users"));
589    }
590
591    #[test]
592    fn manual_start_lsn_is_rejected() {
593        let mut config = connector_config();
594        config.set("start.lsn", "0/1234ABCD");
595        let error = PostgresCdcConfig::from_config(&config).unwrap_err();
596        assert!(error.to_string().contains("start.lsn"));
597    }
598
599    #[test]
600    fn test_from_config_table_include() {
601        let mut config = connector_config();
602        config.set("table.include", "public.users, public.orders");
603
604        let cfg = PostgresCdcConfig::from_config(&config).unwrap();
605        assert_eq!(cfg.table_include, vec!["public.orders", "public.users"]);
606    }
607
608    #[test]
609    fn table_filters_are_trimmed_nonempty_sorted_and_deduplicated_once() {
610        let mut config = connector_config();
611        config.set(
612            "table.include",
613            " public.users,public.orders,, public.users,   ",
614        );
615        config.set(
616            "table.exclude",
617            " public.audit, ,public.archive,public.audit ",
618        );
619
620        let cfg = PostgresCdcConfig::from_config(&config).unwrap();
621        assert_eq!(cfg.table_include, vec!["public.orders", "public.users"]);
622        assert_eq!(cfg.table_exclude, vec!["public.archive", "public.audit"]);
623        assert!(cfg.should_include_table("public.users"));
624        assert!(!cfg.should_include_table("public.audit"));
625        assert!(!cfg.should_include_table("users"));
626    }
627
628    #[test]
629    fn table_filters_reject_unqualified_or_empty_components() {
630        for table in ["users", ".users", "public."] {
631            let mut config = connector_config();
632            config.set("table.include", table);
633            let error = PostgresCdcConfig::from_config(&config).unwrap_err();
634            assert!(error.to_string().contains("schema"), "{table}: {error}");
635        }
636    }
637
638    #[test]
639    fn custom_ca_is_admitted_for_verified_tls() {
640        let mut config = connector_config();
641        config.set("ssl.mode", "verify-full");
642        config.set("ssl.ca.cert.path", "/certs/ca.pem");
643        let parsed = PostgresCdcConfig::from_config(&config).unwrap();
644        assert_eq!(
645            parsed.ssl_ca_cert_path,
646            Some(PathBuf::from("/certs/ca.pem"))
647        );
648    }
649
650    #[test]
651    fn plaintext_rejects_unused_ca_path() {
652        let mut config = connector_config();
653        config.set("ssl.ca.cert.path", "/certs/ca.pem");
654        let error = PostgresCdcConfig::from_config(&config).unwrap_err();
655        assert!(error.to_string().contains("ssl.mode=verify-full"));
656    }
657}
laminar_connectors/postgres/cdc/config.rs

laminar_connectors/postgres/cdc/
config.rs
