laminar_connectors/postgres/cdc/config/
mod.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;
laminar_connectors/postgres/cdc/config/mod.rs

laminar_connectors/postgres/cdc/config/
mod.rs
