faucet_source_mysql_cdc/
config.rs

1//! Configuration for [`MysqlCdcSource`](crate::MysqlCdcSourceConfig).
2
3use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError};
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6use std::fmt;
7use std::time::Duration;
8
9fn default_idle_timeout() -> Duration {
10    Duration::from_secs(30)
11}
12fn default_batch_size() -> usize {
13    DEFAULT_BATCH_SIZE
14}
15fn default_include_columns() -> bool {
16    true
17}
18
19/// Where to start reading the binlog on a fresh run (no persisted bookmark).
20#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
21#[serde(tag = "type", rename_all = "snake_case")]
22pub enum StartPosition {
23    /// Start at the server's current binlog position (skip history). Default.
24    #[default]
25    Current,
26    /// Start at the earliest available binlog (errors if purged at open).
27    Earliest,
28    /// Start after a specific executed GTID set.
29    GtidSet { value: String },
30    /// Start at an explicit binlog file + position.
31    FilePos { file: String, pos: u64 },
32}
33
34/// TLS configuration for the binlog replication connection.
35#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
36#[serde(tag = "mode", rename_all = "snake_case")]
37pub enum CdcTls {
38    /// No TLS (default, back-compatible). Credentials + row data travel cleartext.
39    #[default]
40    Disable,
41    /// Require TLS but do not verify the server certificate.
42    Require,
43    /// Require TLS and verify the chain against `ca_path` (or system roots).
44    VerifyCa {
45        #[serde(default, skip_serializing_if = "Option::is_none")]
46        ca_path: Option<String>,
47    },
48    /// Require TLS and verify both the chain and the hostname.
49    VerifyFull {
50        #[serde(default, skip_serializing_if = "Option::is_none")]
51        ca_path: Option<String>,
52    },
53}
54
55/// Configuration for the MySQL CDC (binlog) source.
56#[derive(Clone, Serialize, Deserialize, JsonSchema)]
57pub struct MysqlCdcSourceConfig {
58    /// Connection URL, e.g. `mysql://repl:pass@host:3306/db`.
59    pub connection_url: String,
60    /// Replica server id. **Required** and must be unique across all replicas.
61    pub server_id: u32,
62    /// Optional client-side allowlist of `db.table` names. Empty = all tables.
63    #[serde(default)]
64    pub include_tables: Vec<String>,
65    /// Optional client-side blocklist of `db.table` names (allowlist wins).
66    #[serde(default)]
67    pub exclude_tables: Vec<String>,
68    /// Start position on a fresh run (ignored once a bookmark exists).
69    #[serde(default)]
70    pub start_position: StartPosition,
71    /// Emit the pre-image (`before`) on updates/deletes. Default true.
72    #[serde(default = "default_include_columns")]
73    pub include_columns: bool,
74    /// Emit DDL statements as `{op:"ddl"}` records. Default false.
75    #[serde(default)]
76    pub emit_schema_changes: bool,
77    /// TLS settings for the replication connection. Default `disable`.
78    #[serde(default)]
79    pub tls: CdcTls,
80    /// Terminator: end the fetch cycle after this much quiet. Default 30s.
81    #[serde(
82        default = "default_idle_timeout",
83        with = "faucet_core::config::duration_secs"
84    )]
85    #[schemars(with = "u64")]
86    pub idle_timeout: Duration,
87    /// Max records buffered for a single in-progress transaction before abort.
88    /// `None` = unbounded.
89    #[serde(default)]
90    pub max_staged_records: Option<usize>,
91    /// Advisory per-page record count. `0` = accumulate all txns into one page.
92    #[serde(default = "default_batch_size")]
93    pub batch_size: usize,
94}
95
96impl MysqlCdcSourceConfig {
97    /// Validate fail-fast invariants. Called from `MysqlCdcSource::new`.
98    pub fn validate(&self) -> Result<(), FaucetError> {
99        if self.connection_url.trim().is_empty() {
100            return Err(FaucetError::Config(
101                "mysql-cdc: connection_url must not be empty".into(),
102            ));
103        }
104        if self.server_id == 0 {
105            return Err(FaucetError::Config(
106                "mysql-cdc: server_id must be a non-zero value unique across replicas".into(),
107            ));
108        }
109        if self.idle_timeout.is_zero() {
110            return Err(FaucetError::Config(
111                "mysql-cdc: idle_timeout must be > 0".into(),
112            ));
113        }
114        for t in self.include_tables.iter().chain(self.exclude_tables.iter()) {
115            if !t.contains('.') {
116                return Err(FaucetError::Config(format!(
117                    "mysql-cdc: table filter '{t}' must be fully qualified as 'database.table'"
118                )));
119            }
120        }
121        faucet_core::validate_batch_size(self.batch_size)?;
122        let key = crate::state::state_key(self.server_id);
123        faucet_core::state::validate_state_key(&key)?;
124        Ok(())
125    }
126
127    /// Whether a `db.table` passes the include/exclude filters (allowlist wins).
128    pub fn table_included(&self, db: &str, table: &str) -> bool {
129        let q = format!("{db}.{table}");
130        if !self.include_tables.is_empty() {
131            return self.include_tables.iter().any(|t| t == &q);
132        }
133        !self.exclude_tables.iter().any(|t| t == &q)
134    }
135}
136
137impl fmt::Debug for MysqlCdcSourceConfig {
138    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
139        f.debug_struct("MysqlCdcSourceConfig")
140            .field("connection_url", &"***")
141            .field("server_id", &self.server_id)
142            .field("include_tables", &self.include_tables)
143            .field("exclude_tables", &self.exclude_tables)
144            .field("start_position", &self.start_position)
145            .field("include_columns", &self.include_columns)
146            .field("emit_schema_changes", &self.emit_schema_changes)
147            .field("tls", &self.tls)
148            .field("idle_timeout", &self.idle_timeout)
149            .field("max_staged_records", &self.max_staged_records)
150            .field("batch_size", &self.batch_size)
151            .finish()
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    use super::*;
158    use serde_json::json;
159
160    fn minimal() -> MysqlCdcSourceConfig {
161        serde_json::from_value(json!({
162            "connection_url": "mysql://repl:pass@localhost:3306/db",
163            "server_id": 1001
164        }))
165        .unwrap()
166    }
167
168    #[test]
169    fn defaults_via_serde() {
170        let c = minimal();
171        assert_eq!(c.idle_timeout.as_secs(), 30);
172        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
173        assert_eq!(c.start_position, StartPosition::Current);
174        assert!(c.include_columns);
175        assert!(!c.emit_schema_changes);
176        assert_eq!(c.tls, CdcTls::Disable);
177    }
178
179    #[test]
180    fn start_position_tagged_enum() {
181        let c: MysqlCdcSourceConfig = serde_json::from_value(json!({
182            "connection_url": "mysql://h/db", "server_id": 5,
183            "start_position": { "type": "file_pos", "file": "binlog.000003", "pos": 4 }
184        }))
185        .unwrap();
186        assert_eq!(
187            c.start_position,
188            StartPosition::FilePos {
189                file: "binlog.000003".into(),
190                pos: 4
191            }
192        );
193    }
194
195    #[test]
196    fn rejects_zero_server_id() {
197        let mut c = minimal();
198        c.server_id = 0;
199        assert!(c.validate().is_err());
200    }
201
202    #[test]
203    fn rejects_unqualified_table_filter() {
204        let mut c = minimal();
205        c.include_tables = vec!["users".into()];
206        assert!(c.validate().is_err());
207    }
208
209    #[test]
210    fn table_filter_allowlist_wins() {
211        let mut c = minimal();
212        c.include_tables = vec!["db.users".into()];
213        c.exclude_tables = vec!["db.users".into()];
214        assert!(c.table_included("db", "users"));
215        assert!(!c.table_included("db", "orders"));
216    }
217
218    #[test]
219    fn table_filter_blocklist() {
220        let mut c = minimal();
221        c.exclude_tables = vec!["db.secret".into()];
222        assert!(c.table_included("db", "users"));
223        assert!(!c.table_included("db", "secret"));
224    }
225
226    #[test]
227    fn debug_redacts_connection_url() {
228        let c: MysqlCdcSourceConfig = serde_json::from_value(json!({
229            "connection_url": "mysql://repl:secret@h:3306/db", "server_id": 1
230        }))
231        .unwrap();
232        let dbg = format!("{c:?}");
233        assert!(dbg.contains("***"));
234        assert!(!dbg.contains("secret"));
235    }
236
237    #[test]
238    fn accepts_minimal() {
239        assert!(minimal().validate().is_ok());
240    }
241}
faucet_source_mysql_cdc/config.rs

faucet_source_mysql_cdc/
config.rs
