datum_cdc/
config.rs

1use std::{
2    sync::Arc,
3    time::{Duration, Instant},
4};
5
6use pgwire_replication::{ReplicationConfig, TlsConfig};
7use url::Url;
8
9use crate::{CdcCheckpointStore, CdcError, CdcResult, PgLsn};
10
11/// PostgreSQL connection settings for `datum-cdc`.
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct PostgresCdcConfig {
14    pub host: String,
15    pub port: u16,
16    pub user: String,
17    pub password: String,
18    pub database: String,
19    pub tls: CdcTlsConfig,
20    pub application_name: String,
21}
22
23impl PostgresCdcConfig {
24    pub fn from_url(value: impl AsRef<str>) -> CdcResult<Self> {
25        let url = Url::parse(value.as_ref())
26            .map_err(|err| CdcError::Config(format!("invalid PostgreSQL URL: {err}")))?;
27        if url.scheme() != "postgresql" && url.scheme() != "postgres" {
28            return Err(CdcError::Config(format!(
29                "unsupported PostgreSQL URL scheme {:?}",
30                url.scheme()
31            )));
32        }
33        let host = url
34            .host_str()
35            .ok_or_else(|| CdcError::Config("PostgreSQL URL is missing a host".into()))?
36            .to_owned();
37        let user = percent_decode(url.username());
38        if user.is_empty() {
39            return Err(CdcError::Config(
40                "PostgreSQL URL is missing a username".into(),
41            ));
42        }
43        let database = url.path().trim_start_matches('/').to_owned();
44        if database.is_empty() {
45            return Err(CdcError::Config(
46                "PostgreSQL URL is missing a database name".into(),
47            ));
48        }
49        let tls = url
50            .query_pairs()
51            .find_map(|(key, value)| (key == "sslmode").then(|| CdcTlsConfig::from_sslmode(&value)))
52            .transpose()?
53            .unwrap_or(CdcTlsConfig::Disable);
54        Ok(Self {
55            host,
56            port: url.port().unwrap_or(5432),
57            user,
58            password: url.password().map(percent_decode).unwrap_or_default(),
59            database,
60            tls,
61            application_name: "datum-cdc".to_owned(),
62        })
63    }
64
65    #[must_use]
66    pub fn replication_config(
67        &self,
68        slot: &str,
69        publication: &str,
70        start_lsn: PgLsn,
71        status_interval: Duration,
72        idle_wakeup_interval: Duration,
73        buffer_events: usize,
74    ) -> ReplicationConfig {
75        ReplicationConfig {
76            host: self.host.clone(),
77            port: self.port,
78            user: self.user.clone(),
79            password: self.password.clone(),
80            database: self.database.clone(),
81            tls: self.tls.to_pgwire(),
82            slot: slot.to_owned(),
83            publication: publication.to_owned(),
84            start_lsn: start_lsn.into(),
85            stop_at_lsn: None,
86            status_interval,
87            idle_wakeup_interval,
88            buffer_events,
89        }
90    }
91}
92
93fn percent_decode(value: &str) -> String {
94    url::form_urlencoded::parse(value.as_bytes())
95        .map(|(decoded, _)| decoded.into_owned())
96        .next()
97        .unwrap_or_else(|| value.to_owned())
98}
99
100/// TLS mode for the replication connection.
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub enum CdcTlsConfig {
103    Disable,
104    Require,
105    VerifyCa(Option<std::path::PathBuf>),
106    VerifyFull(Option<std::path::PathBuf>),
107}
108
109impl CdcTlsConfig {
110    pub fn from_sslmode(value: &str) -> CdcResult<Self> {
111        match value {
112            "disable" => Ok(Self::Disable),
113            "prefer" => Ok(Self::Disable),
114            "require" => Ok(Self::Require),
115            "verify-ca" => Ok(Self::VerifyCa(None)),
116            "verify-full" => Ok(Self::VerifyFull(None)),
117            other => Err(CdcError::Config(format!(
118                "unsupported sslmode {other:?}; expected disable, prefer, require, verify-ca, or verify-full"
119            ))),
120        }
121    }
122
123    #[must_use]
124    pub fn to_pgwire(&self) -> TlsConfig {
125        match self {
126            Self::Disable => TlsConfig::disabled(),
127            Self::Require => TlsConfig::require(),
128            Self::VerifyCa(path) => TlsConfig::verify_ca(path.clone()),
129            Self::VerifyFull(path) => TlsConfig::verify_full(path.clone()),
130        }
131    }
132}
133
134/// Starting point for a materialized CDC source.
135#[derive(Clone, Default)]
136pub enum CdcStart {
137    /// Load a durable checkpoint when configured; otherwise start from the
138    /// slot's confirmed LSN.
139    #[default]
140    CheckpointOrSlot,
141    /// Start from the slot's confirmed LSN.
142    SlotConfirmed,
143    /// Start from an explicit LSN.
144    Lsn(PgLsn),
145}
146
147/// Feedback policy. The MVP intentionally exposes only checkpoint-gated
148/// feedback because immediate feedback would violate at-least-once delivery.
149#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
150pub enum FeedbackMode {
151    #[default]
152    AfterCheckpoint,
153}
154
155/// Logical replication slot lifecycle policy.
156#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
157pub enum SlotLifecycle {
158    #[default]
159    Existing,
160    CreateIfMissing,
161    CreateOwned,
162    Temporary,
163}
164
165impl SlotLifecycle {
166    #[must_use]
167    pub const fn may_create(self) -> bool {
168        matches!(
169            self,
170            Self::CreateIfMissing | Self::CreateOwned | Self::Temporary
171        )
172    }
173
174    #[must_use]
175    pub const fn may_drop(self) -> bool {
176        matches!(self, Self::CreateOwned | Self::Temporary)
177    }
178
179    #[must_use]
180    pub const fn temporary(self) -> bool {
181        matches!(self, Self::Temporary)
182    }
183}
184
185#[derive(Clone)]
186pub(crate) struct SourceSettings {
187    pub(crate) postgres: PostgresCdcConfig,
188    pub(crate) slot: String,
189    pub(crate) publication: String,
190    pub(crate) source: crate::SourceMetadata,
191    pub(crate) start: CdcStart,
192    pub(crate) feedback: FeedbackMode,
193    pub(crate) slot_lifecycle: SlotLifecycle,
194    pub(crate) buffer_capacity: usize,
195    pub(crate) replication_buffer_events: usize,
196    pub(crate) status_interval: Duration,
197    pub(crate) idle_wakeup_interval: Duration,
198    pub(crate) validate_publication: bool,
199    pub(crate) checkpoint_store: Option<Arc<dyn CdcCheckpointStore>>,
200    pub(crate) reconnect: Option<ReconnectSettings>,
201}
202
203impl SourceSettings {
204    pub(crate) fn start_deadline(&self) -> Option<Instant> {
205        self.reconnect
206            .as_ref()
207            .map(|settings| Instant::now() + settings.max_elapsed)
208    }
209}
210
211/// Bounded reconnect policy used by the carrier task.
212#[derive(Debug, Clone)]
213pub struct ReconnectSettings {
214    pub min_backoff: Duration,
215    pub max_backoff: Duration,
216    pub max_elapsed: Duration,
217}
218
219impl Default for ReconnectSettings {
220    fn default() -> Self {
221        Self {
222            min_backoff: Duration::from_millis(200),
223            max_backoff: Duration::from_secs(5),
224            max_elapsed: Duration::from_secs(60),
225        }
226    }
227}
datum_cdc/config.rs

datum_cdc/
config.rs
