faucet_source_mssql_cdc/
config.rs

1//! Configuration for [`MssqlCdcSource`](crate::MssqlCdcSource).
2
3use std::fmt;
4use std::time::Duration;
5
6use faucet_common_mssql::MssqlConnectionConfig;
7use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError, validate_batch_size};
8use schemars::JsonSchema;
9use serde::{Deserialize, Serialize};
10
11fn default_poll_interval() -> Duration {
12    Duration::from_secs(1)
13}
14fn default_idle_timeout() -> Duration {
15    Duration::from_secs(30)
16}
17fn default_batch_size() -> usize {
18    DEFAULT_BATCH_SIZE
19}
20fn default_max_connections() -> u32 {
21    5
22}
23fn default_statement_timeout_secs() -> u64 {
24    300
25}
26
27/// Where to start reading changes on a fresh run (no persisted bookmark).
28#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
29#[serde(tag = "type", rename_all = "snake_case")]
30pub enum StartPosition {
31    /// Start at the database's current maximum LSN — skip all pre-existing
32    /// change history and only capture changes committed after the source
33    /// starts. Default.
34    #[default]
35    Current,
36    /// Start at the earliest LSN still retained by the CDC capture instance
37    /// (`sys.fn_cdc_get_min_lsn`), replaying whatever history the cleanup job
38    /// has not yet purged.
39    Earliest,
40}
41
42/// Configuration for the Microsoft SQL Server CDC source.
43///
44/// The source polls native SQL Server change data capture: `sys.fn_cdc_get_max_lsn()`
45/// for the high-water LSN, then `cdc.fn_cdc_get_all_changes_<capture_instance>()`
46/// per configured capture instance, advancing a durable per-instance LSN bookmark.
47#[derive(Clone, Serialize, Deserialize, JsonSchema)]
48pub struct MssqlCdcSourceConfig {
49    /// Connection + TLS settings (`connection_url` or `connection_string`).
50    #[serde(flatten)]
51    pub connection: MssqlConnectionConfig,
52    /// Capture instances to poll (e.g. `dbo_Orders`). **Required and non-empty.**
53    ///
54    /// A capture instance is created by `sys.sp_cdc_enable_table` and defaults to
55    /// `<schema>_<table>`. Names are used verbatim to build the
56    /// `cdc.fn_cdc_get_all_changes_<name>` table-valued function call, so v1 only
57    /// accepts the SQL Server identifier characters `[A-Za-z0-9_]` (validated at
58    /// load time) to prevent injection into the function name.
59    pub capture_instances: Vec<String>,
60    /// Start position on a fresh run (ignored once a bookmark exists). Default
61    /// `current` (skip existing history).
62    #[serde(default)]
63    pub start_position: StartPosition,
64    /// Seconds to wait between empty polls (no new changes). Default 1s.
65    #[serde(
66        default = "default_poll_interval",
67        with = "faucet_core::config::duration_secs"
68    )]
69    #[schemars(with = "u64")]
70    pub poll_interval: Duration,
71    /// Terminator: end the fetch cycle after this much continuous quiet (no
72    /// change rows across any capture instance). Default 30s. A long-running
73    /// runtime (`faucet schedule` / `faucet serve`) re-invokes the source to
74    /// keep tailing.
75    #[serde(
76        default = "default_idle_timeout",
77        with = "faucet_core::config::duration_secs"
78    )]
79    #[schemars(with = "u64")]
80    pub idle_timeout: Duration,
81    /// Max records buffered for a single in-progress transaction before the run
82    /// aborts with a typed error (rather than risking unbounded memory growth).
83    /// `None` = unbounded.
84    #[serde(default)]
85    pub max_staged_records: Option<usize>,
86    /// Advisory per-page record count. `0` accumulates every change into a
87    /// single trailing page (snapshot/test convenience). Default
88    /// [`DEFAULT_BATCH_SIZE`].
89    #[serde(default = "default_batch_size")]
90    pub batch_size: usize,
91    /// Maximum pooled connections. Defaults to 5.
92    #[serde(default = "default_max_connections")]
93    pub max_connections: u32,
94    /// Per-query timeout in seconds (`0` disables). Defaults to 300.
95    #[serde(default = "default_statement_timeout_secs")]
96    pub statement_timeout_secs: u64,
97    /// Explicit state-store key for the LSN bookmark. When unset, a key is
98    /// derived from the database (or host) and the sorted capture-instance list.
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub state_key: Option<String>,
101}
102
103impl MssqlCdcSourceConfig {
104    /// Validate fail-fast invariants. Called from [`MssqlCdcSource::new`](crate::MssqlCdcSource::new).
105    pub fn validate(&self) -> Result<(), FaucetError> {
106        self.connection.validate()?;
107        if self.capture_instances.is_empty() {
108            return Err(FaucetError::Config(
109                "mssql-cdc: `capture_instances` must list at least one CDC capture instance".into(),
110            ));
111        }
112        for ci in &self.capture_instances {
113            validate_capture_instance(ci)?;
114        }
115        // Duplicate capture instances would double-emit and fight over the same
116        // bookmark map entry — reject them up front.
117        let mut seen = std::collections::BTreeSet::new();
118        for ci in &self.capture_instances {
119            if !seen.insert(ci) {
120                return Err(FaucetError::Config(format!(
121                    "mssql-cdc: duplicate capture instance {ci:?} in `capture_instances`"
122                )));
123            }
124        }
125        if self.poll_interval.is_zero() {
126            return Err(FaucetError::Config(
127                "mssql-cdc: poll_interval must be > 0".into(),
128            ));
129        }
130        if self.idle_timeout.is_zero() {
131            return Err(FaucetError::Config(
132                "mssql-cdc: idle_timeout must be > 0".into(),
133            ));
134        }
135        validate_batch_size(self.batch_size)?;
136        let key = self.resolved_state_key();
137        faucet_core::state::validate_state_key(&key)?;
138        Ok(())
139    }
140
141    /// The state-store key for this source's LSN bookmark map. Uses the explicit
142    /// `state_key` override when set, otherwise a derived, stable key.
143    pub fn resolved_state_key(&self) -> String {
144        self.state_key
145            .clone()
146            .unwrap_or_else(|| derive_state_key(self))
147    }
148}
149
150/// Validate a capture-instance name: non-empty, ≤128 chars, and only the SQL
151/// Server identifier characters `[A-Za-z0-9_]`. The name is interpolated into
152/// the `cdc.fn_cdc_get_all_changes_<name>` function identifier (which cannot be
153/// bracket-quoted mid-name), so this is the injection guard.
154pub(crate) fn validate_capture_instance(ci: &str) -> Result<(), FaucetError> {
155    if ci.is_empty() {
156        return Err(FaucetError::Config(
157            "mssql-cdc: capture instance name must not be empty".into(),
158        ));
159    }
160    if ci.len() > 128 {
161        return Err(FaucetError::Config(format!(
162            "mssql-cdc: capture instance name {ci:?} exceeds 128 characters"
163        )));
164    }
165    if !ci.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
166        return Err(FaucetError::Config(format!(
167            "mssql-cdc: capture instance name {ci:?} may only contain letters, digits, and \
168             underscores (SQL Server identifier characters); other characters are unsupported in v1"
169        )));
170    }
171    Ok(())
172}
173
174/// Derive a stable state key from the target database (or connection host) plus
175/// a fingerprint of the sorted capture-instance list. Pure so it is unit-testable.
176///
177/// - Single instance: `mssql-cdc:<db-or-host>:<capture_instance>` (readable).
178/// - Multiple instances: `mssql-cdc:<db-or-host>:<fnv1a-hex>` (stable digest).
179pub(crate) fn derive_state_key(config: &MssqlCdcSourceConfig) -> String {
180    let scope = database_or_host(config);
181
182    if config.capture_instances.len() == 1 {
183        return format!("mssql-cdc:{scope}:{}", config.capture_instances[0]);
184    }
185
186    let mut sorted: Vec<&str> = config
187        .capture_instances
188        .iter()
189        .map(String::as_str)
190        .collect();
191    sorted.sort_unstable();
192    let digest = fnv1a_hex(&sorted.join(","));
193    format!("mssql-cdc:{scope}:{digest}")
194}
195
196/// Extract the target database name (preferred) or connection host for the state
197/// key scope, sanitising to key-safe characters. Falls back to `mssql`.
198fn database_or_host(config: &MssqlCdcSourceConfig) -> String {
199    let raw = config
200        .connection
201        .connection_url
202        .as_deref()
203        .and_then(|u| url::Url::parse(u).ok())
204        .and_then(|u| {
205            let db = u.path().trim_start_matches('/').to_string();
206            if !db.is_empty() {
207                Some(db)
208            } else {
209                u.host_str().map(str::to_string)
210            }
211        })
212        .or_else(|| {
213            config
214                .connection
215                .connection_string
216                .as_deref()
217                .and_then(database_from_ado_string)
218        })
219        .unwrap_or_else(|| "mssql".to_string());
220
221    let sanitised: String = raw
222        .chars()
223        .map(|c| {
224            if c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.') {
225                c
226            } else {
227                '_'
228            }
229        })
230        .collect();
231    if sanitised.is_empty() {
232        "mssql".to_string()
233    } else {
234        sanitised
235    }
236}
237
238/// Pull the `Database=`/`Initial Catalog=` value out of an ADO.NET connection
239/// string (case-insensitive key match). Pure.
240fn database_from_ado_string(s: &str) -> Option<String> {
241    for part in s.split(';') {
242        let mut kv = part.splitn(2, '=');
243        let key = kv.next()?.trim();
244        let val = kv.next().unwrap_or("").trim();
245        if key.eq_ignore_ascii_case("database") || key.eq_ignore_ascii_case("initial catalog") {
246            if val.is_empty() {
247                return None;
248            }
249            return Some(val.to_string());
250        }
251    }
252    None
253}
254
255/// 64-bit FNV-1a of `s`, rendered as 16 lowercase hex digits. Deterministic and
256/// dependency-free (used only for a stable state-key digest, not security).
257fn fnv1a_hex(s: &str) -> String {
258    const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
259    const PRIME: u64 = 0x0000_0100_0000_01b3;
260    let mut hash = OFFSET;
261    for b in s.as_bytes() {
262        hash ^= u64::from(*b);
263        hash = hash.wrapping_mul(PRIME);
264    }
265    format!("{hash:016x}")
266}
267
268impl fmt::Debug for MssqlCdcSourceConfig {
269    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
270        f.debug_struct("MssqlCdcSourceConfig")
271            .field("connection", &"***")
272            .field("capture_instances", &self.capture_instances)
273            .field("start_position", &self.start_position)
274            .field("poll_interval", &self.poll_interval)
275            .field("idle_timeout", &self.idle_timeout)
276            .field("max_staged_records", &self.max_staged_records)
277            .field("batch_size", &self.batch_size)
278            .field("max_connections", &self.max_connections)
279            .field("statement_timeout_secs", &self.statement_timeout_secs)
280            .field("state_key", &self.state_key)
281            .finish()
282    }
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288    use serde_json::json;
289
290    fn minimal() -> MssqlCdcSourceConfig {
291        serde_json::from_value(json!({
292            "connection_url": "mssql://sa:pw@localhost:1433/sales",
293            "capture_instances": ["dbo_Orders"]
294        }))
295        .unwrap()
296    }
297
298    #[test]
299    fn defaults_via_serde() {
300        let c = minimal();
301        assert_eq!(c.poll_interval.as_secs(), 1);
302        assert_eq!(c.idle_timeout.as_secs(), 30);
303        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
304        assert_eq!(c.max_connections, 5);
305        assert_eq!(c.statement_timeout_secs, 300);
306        assert_eq!(c.start_position, StartPosition::Current);
307        assert!(c.max_staged_records.is_none());
308    }
309
310    #[test]
311    fn start_position_tagged_enum() {
312        let c: MssqlCdcSourceConfig = serde_json::from_value(json!({
313            "connection_url": "mssql://sa:pw@h/db",
314            "capture_instances": ["dbo_t"],
315            "start_position": { "type": "earliest" }
316        }))
317        .unwrap();
318        assert_eq!(c.start_position, StartPosition::Earliest);
319    }
320
321    #[test]
322    fn accepts_minimal() {
323        assert!(minimal().validate().is_ok());
324    }
325
326    #[test]
327    fn rejects_empty_capture_instances() {
328        let c: MssqlCdcSourceConfig = serde_json::from_value(json!({
329            "connection_url": "mssql://sa:pw@h/db",
330            "capture_instances": []
331        }))
332        .unwrap();
333        assert!(c.validate().is_err());
334    }
335
336    #[test]
337    fn rejects_duplicate_capture_instances() {
338        let mut c = minimal();
339        c.capture_instances = vec!["dbo_t".into(), "dbo_t".into()];
340        assert!(c.validate().is_err());
341    }
342
343    #[test]
344    fn rejects_injection_in_capture_instance() {
345        assert!(validate_capture_instance("dbo_Orders").is_ok());
346        assert!(validate_capture_instance("Orders123").is_ok());
347        // Hostile names that would break out of the function identifier.
348        assert!(validate_capture_instance("dbo.Orders").is_err());
349        assert!(validate_capture_instance("x(1,2,'all')--").is_err());
350        assert!(validate_capture_instance("a b").is_err());
351        assert!(validate_capture_instance("").is_err());
352        assert!(validate_capture_instance(&"a".repeat(129)).is_err());
353    }
354
355    #[test]
356    fn rejects_zero_intervals() {
357        let mut c = minimal();
358        c.poll_interval = Duration::ZERO;
359        assert!(c.validate().is_err());
360
361        let mut c = minimal();
362        c.idle_timeout = Duration::ZERO;
363        assert!(c.validate().is_err());
364    }
365
366    #[test]
367    fn rejects_bad_batch_size() {
368        let mut c = minimal();
369        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
370        assert!(c.validate().is_err());
371    }
372
373    #[test]
374    fn requires_a_connection_source() {
375        let c: MssqlCdcSourceConfig = serde_json::from_value(json!({
376            "capture_instances": ["dbo_t"]
377        }))
378        .unwrap();
379        // No connection_url / connection_string -> connection.validate() fails.
380        assert!(c.validate().is_err());
381    }
382
383    #[test]
384    fn state_key_single_instance_is_readable() {
385        let c = minimal();
386        let key = c.resolved_state_key();
387        assert_eq!(key, "mssql-cdc:sales:dbo_Orders");
388        faucet_core::state::validate_state_key(&key).expect("derived key must be valid");
389    }
390
391    #[test]
392    fn state_key_multi_instance_is_digest_and_order_independent() {
393        let mut a = minimal();
394        a.capture_instances = vec!["dbo_Orders".into(), "dbo_Items".into()];
395        let mut b = minimal();
396        b.capture_instances = vec!["dbo_Items".into(), "dbo_Orders".into()];
397        let ka = a.resolved_state_key();
398        let kb = b.resolved_state_key();
399        assert_eq!(ka, kb, "sorted digest is order-independent");
400        assert!(ka.starts_with("mssql-cdc:sales:"));
401        assert!(!ka.ends_with("dbo_Orders"));
402        faucet_core::state::validate_state_key(&ka).expect("digest key must be valid");
403    }
404
405    #[test]
406    fn state_key_explicit_override_wins() {
407        let mut c = minimal();
408        c.state_key = Some("custom:key".into());
409        assert_eq!(c.resolved_state_key(), "custom:key");
410    }
411
412    #[test]
413    fn state_key_falls_back_to_host_then_mssql() {
414        // connection_string with a database.
415        let c: MssqlCdcSourceConfig = serde_json::from_value(json!({
416            "connection_string": "Server=tcp:h,1433;Initial Catalog=warehouse;User Id=sa;Password=p",
417            "capture_instances": ["dbo_t"]
418        }))
419        .unwrap();
420        assert_eq!(c.resolved_state_key(), "mssql-cdc:warehouse:dbo_t");
421
422        // connection_string without a database -> "mssql".
423        let c: MssqlCdcSourceConfig = serde_json::from_value(json!({
424            "connection_string": "Server=tcp:h,1433;User Id=sa;Password=p",
425            "capture_instances": ["dbo_t"]
426        }))
427        .unwrap();
428        assert_eq!(c.resolved_state_key(), "mssql-cdc:mssql:dbo_t");
429    }
430
431    #[test]
432    fn debug_redacts_connection() {
433        let c: MssqlCdcSourceConfig = serde_json::from_value(json!({
434            "connection_url": "mssql://sa:secret@h/db",
435            "capture_instances": ["dbo_t"]
436        }))
437        .unwrap();
438        let dbg = format!("{c:?}");
439        assert!(dbg.contains("***"));
440        assert!(!dbg.contains("secret"));
441    }
442}
faucet_source_mssql_cdc/config.rs

faucet_source_mssql_cdc/
config.rs
