faucet_source_oracle_cdc/
config.rs

1//! Configuration for [`OracleCdcSource`](crate::OracleCdcSource).
2
3use std::time::Duration;
4
5use faucet_common_oracle::{OracleConnectionConfig, quote_ident_oracle, split_table};
6use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError, validate_batch_size};
7use schemars::JsonSchema;
8use serde::{Deserialize, Serialize};
9
10fn default_poll_interval() -> Duration {
11    Duration::from_secs(1)
12}
13fn default_idle_timeout() -> Duration {
14    Duration::from_secs(30)
15}
16fn default_batch_size() -> usize {
17    DEFAULT_BATCH_SIZE
18}
19fn default_max_connections() -> u32 {
20    2
21}
22fn default_statement_timeout_secs() -> u64 {
23    600
24}
25fn default_max_scn_window() -> u64 {
26    500_000
27}
28fn default_flush_table() -> String {
29    "FAUCET_LOGMNR_FLUSH".into()
30}
31
32/// Where to start mining on a fresh run (no persisted bookmark).
33#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
34#[serde(tag = "type", rename_all = "snake_case")]
35pub enum StartPosition {
36    /// The database's current SCN — only changes committed from now on
37    /// (transactions already open are captured from their start). Default.
38    #[default]
39    Current,
40    /// The oldest SCN still present in the available redo (online + archived).
41    Earliest,
42}
43
44/// What to do with a change LogMiner cannot render as SQL for a tracked table
45/// (`UNSUPPORTED` rows, or a *dictionary mismatch* after the table's structure
46/// changed since the redo was written).
47#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
48#[serde(rename_all = "snake_case")]
49pub enum OnUnsupported {
50    /// Fail the run (default) — never silently drop a change.
51    #[default]
52    Fail,
53    /// Log a warning and skip the change.
54    Skip,
55}
56
57/// Configuration for the Oracle LogMiner CDC source.
58#[derive(Clone, Serialize, Deserialize, JsonSchema)]
59#[schemars(extend("x-faucet-follows-servers" = true))]
60pub struct OracleCdcSourceConfig {
61    /// Connection settings. Connect to the PDB that owns the tables (Oracle
62    /// 21c+ mines per PDB), or to a non-CDB / the CDB root.
63    #[serde(flatten)]
64    pub connection: OracleConnectionConfig,
65    /// Tables to capture as `OWNER.TABLE`, in data-dictionary case (usually
66    /// upper case). **Required and non-empty.**
67    pub tables: Vec<String>,
68    /// Start position on a fresh run. Default `current`.
69    #[serde(default)]
70    pub start_position: StartPosition,
71    /// Seconds to wait between polls that find nothing new. Default 1s.
72    #[serde(
73        default = "default_poll_interval",
74        with = "faucet_core::config::duration_secs"
75    )]
76    #[schemars(with = "u64")]
77    pub poll_interval: Duration,
78    /// End the fetch cycle after this long without a captured change, counted
79    /// only once mining has reached the current SCN. Default 30s. A
80    /// long-running runtime re-invokes the source to keep tailing.
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    /// Largest SCN range mined per LogMiner session. Default 500000.
88    #[serde(default = "default_max_scn_window")]
89    pub max_scn_window: u64,
90    /// Abort when one in-flight transaction buffers more than this many
91    /// changes (bounded memory). Defaults to 1,000,000; `null` = unbounded.
92    #[serde(default = "faucet_core::replication::default_max_staged_records")]
93    #[schemars(extend("x-faucet-caller-max" = 1_000_000))]
94    pub max_staged_records: Option<usize>,
95    /// Abort when all open transactions together buffer more than this many
96    /// changes, counting LOB pieces and folded updates. Unset or `null` =
97    /// unbounded.
98    #[serde(default)]
99    #[schemars(extend("x-faucet-caller-max" = 1_000_000, "x-faucet-caller-set" = true))]
100    pub max_staged_records_total: Option<usize>,
101    /// `0` aggregates committed transactions into pages of up to 100,000
102    /// records, emitted at least every `idle_timeout`, each with its bookmark;
103    /// otherwise each committed transaction is its own page. Default
104    /// [`DEFAULT_BATCH_SIZE`].
105    #[serde(default = "default_batch_size")]
106    pub batch_size: usize,
107    /// Maximum pooled sessions. Default 2.
108    #[serde(default = "default_max_connections")]
109    pub max_connections: u32,
110    /// Per-round-trip call timeout in seconds (`0` disables). Default 600.
111    #[serde(default = "default_statement_timeout_secs")]
112    pub statement_timeout_secs: u64,
113    /// Table the source commits to before each mining window, forcing the log
114    /// writer to flush redo up to the window's end SCN. Created in the
115    /// connecting user's schema when missing. Default `FAUCET_LOGMNR_FLUSH`.
116    #[serde(default = "default_flush_table")]
117    pub flush_table: String,
118    /// Handling of changes LogMiner cannot render. Default `fail`.
119    #[serde(default)]
120    pub on_unsupported: OnUnsupported,
121    /// Explicit state-store key. Derived as `oracle-cdc:<database>:<table>`
122    /// (or a digest of several tables) when unset.
123    #[serde(default, skip_serializing_if = "Option::is_none")]
124    pub state_key: Option<String>,
125}
126
127impl std::fmt::Debug for OracleCdcSourceConfig {
128    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
129        f.debug_struct("OracleCdcSourceConfig")
130            .field("connection", &self.connection)
131            .field("tables", &self.tables)
132            .field("start_position", &self.start_position)
133            .field("poll_interval", &self.poll_interval)
134            .field("idle_timeout", &self.idle_timeout)
135            .field("max_scn_window", &self.max_scn_window)
136            .field("max_staged_records", &self.max_staged_records)
137            .field("max_staged_records_total", &self.max_staged_records_total)
138            .field("batch_size", &self.batch_size)
139            .field("max_connections", &self.max_connections)
140            .field("statement_timeout_secs", &self.statement_timeout_secs)
141            .field("flush_table", &self.flush_table)
142            .field("on_unsupported", &self.on_unsupported)
143            .field("state_key", &self.state_key)
144            .finish()
145    }
146}
147
148impl OracleCdcSourceConfig {
149    /// A config capturing `tables` with defaults elsewhere.
150    pub fn new(connection: OracleConnectionConfig, tables: Vec<String>) -> Self {
151        Self {
152            connection,
153            tables,
154            start_position: StartPosition::default(),
155            poll_interval: default_poll_interval(),
156            idle_timeout: default_idle_timeout(),
157            max_scn_window: default_max_scn_window(),
158            max_staged_records: faucet_core::replication::default_max_staged_records(),
159            max_staged_records_total: None,
160            batch_size: default_batch_size(),
161            max_connections: default_max_connections(),
162            statement_timeout_secs: default_statement_timeout_secs(),
163            flush_table: default_flush_table(),
164            on_unsupported: OnUnsupported::default(),
165            state_key: None,
166        }
167    }
168
169    /// Validate the fail-fast invariants.
170    pub fn validate(&self) -> Result<(), FaucetError> {
171        self.connection.validate()?;
172        if self.tables.is_empty() {
173            return Err(FaucetError::Config(
174                "oracle-cdc: `tables` must list at least one OWNER.TABLE".into(),
175            ));
176        }
177        let mut seen = std::collections::BTreeSet::new();
178        for t in &self.tables {
179            let (owner, name) = split_table(t)?;
180            let Some(owner) = owner else {
181                return Err(FaucetError::Config(format!(
182                    "oracle-cdc: table {t:?} must be owner-qualified (OWNER.TABLE)"
183                )));
184            };
185            quote_ident_oracle(&owner)?;
186            quote_ident_oracle(&name)?;
187            if !seen.insert(t) {
188                return Err(FaucetError::Config(format!(
189                    "oracle-cdc: duplicate table {t:?} in `tables`"
190                )));
191            }
192        }
193        quote_ident_oracle(&self.flush_table)?;
194        if self.poll_interval.is_zero() || self.idle_timeout.is_zero() {
195            return Err(FaucetError::Config(
196                "oracle-cdc: poll_interval and idle_timeout must be > 0".into(),
197            ));
198        }
199        if self.max_scn_window == 0 {
200            return Err(FaucetError::Config(
201                "oracle-cdc: max_scn_window must be > 0".into(),
202            ));
203        }
204        validate_batch_size(self.batch_size)?;
205        faucet_core::state::validate_state_key(&self.resolved_state_key())?;
206        Ok(())
207    }
208
209    /// `(owner, table)` pairs of the captured tables.
210    pub fn table_pairs(&self) -> Vec<(String, String)> {
211        self.tables
212            .iter()
213            .filter_map(|t| match split_table(t) {
214                Ok((Some(o), n)) => Some((o, n)),
215                _ => None,
216            })
217            .collect()
218    }
219
220    /// The state-store key for the SCN bookmark.
221    pub fn resolved_state_key(&self) -> String {
222        if let Some(k) = &self.state_key {
223            return k.clone();
224        }
225        self.key_for_scope(self.connection.scope_label())
226    }
227
228    /// Keys the SCN bookmark was stored under by releases whose scope did not
229    /// name the host and port; empty when the key is configured explicitly.
230    pub fn legacy_state_keys(&self) -> Vec<String> {
231        if self.state_key.is_some() {
232            return Vec::new();
233        }
234        self.connection
235            .legacy_scope_label()
236            .map(|scope| vec![self.key_for_scope(scope)])
237            .unwrap_or_default()
238    }
239
240    fn key_for_scope(&self, scope: String) -> String {
241        let scope = if scope.is_empty() {
242            "oracle".into()
243        } else {
244            scope
245        };
246        if self.tables.len() == 1 {
247            let t: String = self.tables[0]
248                .chars()
249                .map(|c| {
250                    if c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.') {
251                        c
252                    } else {
253                        '_'
254                    }
255                })
256                .collect();
257            return format!("oracle-cdc:{scope}:{t}");
258        }
259        let mut sorted: Vec<&str> = self.tables.iter().map(String::as_str).collect();
260        sorted.sort_unstable();
261        format!("oracle-cdc:{scope}:{}", fnv1a_hex(&sorted.join(",")))
262    }
263}
264
265fn fnv1a_hex(s: &str) -> String {
266    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
267    for b in s.bytes() {
268        hash ^= u64::from(b);
269        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
270    }
271    format!("{hash:016x}")
272}
273
274#[cfg(test)]
275mod tests;
faucet_source_oracle_cdc/config.rs

faucet_source_oracle_cdc/
config.rs
