laminar_connectors/nats/config/
mod.rs

1//! NATS source and sink configuration.
2
3use std::str::FromStr;
4use std::time::Duration;
5
6use async_nats::HeaderName;
7
8use crate::config::ConnectorConfig;
9use crate::error::ConnectorError;
10use crate::serde::Format;
11
12/// Keep retained acknowledgement futures below async-nats' internal publish
13/// window and put a fixed ceiling on connector memory.
14pub(super) const MAX_PENDING_PUBLISH_ACKS: usize = 4_096;
15
16/// A connector operation must always have a finite client-side bound. This is
17/// deliberately an implementation ceiling rather than another deployment knob.
18pub(super) const MAX_NATS_IO_TIMEOUT: Duration = Duration::from_secs(300);
19
20const NATS_CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
21const NATS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
22
23/// NATS connector mode.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
25pub enum Mode {
26    /// Plain NATS pub/sub. No durability, no replay, at-most-once.
27    Core,
28    /// `JetStream` with durable pull consumers.
29    #[default]
30    JetStream,
31}
32
33str_enum!(fromstr Mode, lowercase, ConnectorError, "invalid nats mode",
34    Core => "core";
35    JetStream => "jetstream"
36);
37
38/// `JetStream` consumer ack policy.
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
40pub enum AckPolicy {
41    /// Each message acked individually.
42    #[default]
43    Explicit,
44    /// No ack required.
45    None,
46}
47
48str_enum!(fromstr AckPolicy, lowercase, ConnectorError, "invalid ack.policy",
49    Explicit => "explicit";
50    None => "none"
51);
52
53/// `JetStream` consumer delivery policy — where to start.
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
55pub enum DeliverPolicy {
56    /// Every message retained by the stream.
57    #[default]
58    All,
59    /// Only messages published after consumer creation.
60    New,
61    /// Start from a user-supplied stream sequence.
62    ByStartSequence,
63    /// Start from a user-supplied timestamp.
64    ByStartTime,
65}
66
67str_enum!(fromstr DeliverPolicy, lowercase, ConnectorError, "invalid deliver.policy",
68    All => "all";
69    New => "new";
70    ByStartSequence => "by_start_sequence";
71    ByStartTime => "by_start_time"
72);
73
74/// Where the sink gets the subject to publish to.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub enum SubjectSpec {
77    /// A single literal subject used for every row.
78    Literal(String),
79    /// Per-row: read the named column and use its string value.
80    Column(String),
81}
82
83/// NATS user-authentication mode. Custom `Debug` redacts secrets
84/// because the parent config structs derive `Debug`.
85#[derive(Clone, Default, PartialEq, Eq)]
86#[allow(missing_docs)]
87pub enum AuthMode {
88    #[default]
89    None,
90    UserPass {
91        user: String,
92        password: String,
93    },
94    Token(String),
95    CredsFile(String),
96}
97
98impl std::fmt::Debug for AuthMode {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        match self {
101            Self::None => f.write_str("None"),
102            Self::UserPass { user, .. } => f
103                .debug_struct("UserPass")
104                .field("user", user)
105                .field("password", &"<redacted>")
106                .finish(),
107            Self::Token(_) => f.debug_tuple("Token").field(&"<redacted>").finish(),
108            Self::CredsFile(path) => f.debug_tuple("CredsFile").field(path).finish(),
109        }
110    }
111}
112
113/// TLS transport configuration.
114#[derive(Debug, Clone, Default)]
115#[allow(missing_docs)]
116pub struct TlsConfig {
117    pub enabled: bool,
118    pub ca_location: Option<String>,
119    pub cert_location: Option<String>,
120    pub key_location: Option<String>,
121}
122
123/// NATS source configuration.
124#[derive(Debug, Clone)]
125#[allow(missing_docs)] // one-line `///` per field noises up the config struct
126pub struct NatsSourceConfig {
127    pub servers: Vec<String>,
128    pub mode: Mode,
129    pub auth: AuthMode,
130    pub tls: TlsConfig,
131
132    // JetStream
133    pub stream: Option<String>,
134    pub subject: Option<String>,
135    pub subject_filters: Vec<String>,
136    pub consumer: Option<String>,
137    pub deliver_policy: DeliverPolicy,
138    pub start_sequence: Option<u64>,
139    pub start_time: Option<String>, // RFC3339; parsed against the NATS client's time type in open()
140    pub ack_policy: AckPolicy,
141    pub ack_wait: Duration,
142    pub max_deliver: i64,
143    pub max_ack_pending: i64,
144    pub fetch_batch: usize,
145    pub fetch_max_wait: Duration,
146    /// Interval between `consumer.info()` polls that feed the
147    /// `nats_source_consumer_lag` gauge. Zero disables the poll.
148    pub lag_poll_interval: Duration,
149
150    // Core
151    pub queue_group: Option<String>,
152
153    pub format: Format,
154}
155
156impl NatsSourceConfig {
157    /// # Errors
158    ///
159    /// `ConfigurationError` on any parse or validation failure.
160    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
161        let servers = parse_servers(config)?;
162        let mode = parse_or_default::<Mode>(config, "mode")?;
163        let auth = parse_auth(config)?;
164        let tls = parse_tls(config)?;
165
166        let stream = config.get("stream").map(str::to_string);
167        let subject = config.get("subject").map(str::to_string);
168        let subject_filters = config
169            .get("subject.filters")
170            .map(split_csv)
171            .unwrap_or_default();
172        let consumer = config.get("consumer").map(str::to_string);
173        let deliver_policy = parse_or_default::<DeliverPolicy>(config, "deliver.policy")?;
174        let start_sequence = parse_opt_u64(config, "start.sequence")?;
175        let start_time = config.get("start.time").map(str::to_string);
176        let ack_policy = parse_or_default::<AckPolicy>(config, "ack.policy")?;
177        let ack_wait = require_positive_duration(config, "ack.wait.ms", Duration::from_secs(60))?;
178        let max_deliver = require_positive_or_unlimited_i64(config, "max.deliver", 5)?;
179        let max_ack_pending = require_positive_or_unlimited_i64(config, "max.ack.pending", 10_000)?;
180        let fetch_batch = require_positive_usize(config, "fetch.batch", 500)?;
181        let fetch_max_wait =
182            require_positive_duration(config, "fetch.max.wait.ms", Duration::from_millis(500))?;
183        let lag_poll_interval =
184            parse_duration_ms(config, "lag.poll.interval.ms", Duration::from_secs(10))?;
185        let queue_group = config.get("queue.group").map(str::to_string);
186        let format = parse_format(config)?;
187
188        let cfg = Self {
189            servers,
190            mode,
191            auth,
192            tls,
193            stream,
194            subject,
195            subject_filters,
196            consumer,
197            deliver_policy,
198            start_sequence,
199            start_time,
200            ack_policy,
201            ack_wait,
202            max_deliver,
203            max_ack_pending,
204            fetch_batch,
205            fetch_max_wait,
206            lag_poll_interval,
207            queue_group,
208            format,
209        };
210        cfg.validate()?;
211        Ok(cfg)
212    }
213
214    fn validate(&self) -> Result<(), ConnectorError> {
215        match self.mode {
216            Mode::JetStream => {
217                if self.stream.is_none() {
218                    return Err(cfg_err(
219                        "[LDB-5040] jetstream mode requires 'stream' to be set",
220                    ));
221                }
222                if self.consumer.is_none() {
223                    return Err(cfg_err(
224                        "[LDB-5041] jetstream mode requires 'consumer' (durable name) — \
225                         ephemeral consumers are not supported",
226                    ));
227                }
228                if self.subject.is_none() && self.subject_filters.is_empty() {
229                    return Err(cfg_err(
230                        "[LDB-5042] jetstream mode requires 'subject' or 'subject.filters' \
231                         — the consumer must have at least one filter",
232                    ));
233                }
234                if matches!(self.deliver_policy, DeliverPolicy::ByStartSequence)
235                    && self.start_sequence.is_none()
236                {
237                    return Err(cfg_err(
238                        "[LDB-5043] deliver.policy=by_start_sequence requires 'start.sequence'",
239                    ));
240                }
241                if matches!(self.deliver_policy, DeliverPolicy::ByStartTime)
242                    && self.start_time.is_none()
243                {
244                    return Err(cfg_err(
245                        "[LDB-5044] deliver.policy=by_start_time requires 'start.time'",
246                    ));
247                }
248            }
249            Mode::Core => {
250                if self.subject.is_none() {
251                    return Err(cfg_err(
252                        "[LDB-5045] core mode requires 'subject' (JetStream stream config \
253                         does not apply)",
254                    ));
255                }
256            }
257        }
258        Ok(())
259    }
260}
261
262/// NATS sink configuration.
263#[derive(Debug, Clone)]
264#[allow(missing_docs)]
265pub struct NatsSinkConfig {
266    pub servers: Vec<String>,
267    pub mode: Mode,
268    pub auth: AuthMode,
269    pub tls: TlsConfig,
270    pub stream: Option<String>,
271    pub subject: SubjectSpec,
272    pub dedup_id_column: Option<String>,
273    /// Stream's `duplicate_window` must be at least this long when bounded
274    /// `Nats-Msg-Id` deduplication is enabled.
275    pub min_duplicate_window: Duration,
276    pub max_pending: usize,
277    pub ack_timeout: Duration,
278    pub format: Format,
279    pub header_columns: Vec<HeaderName>,
280}
281
282impl NatsSinkConfig {
283    /// # Errors
284    ///
285    /// `ConfigurationError` on any parse or validation failure.
286    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
287        let servers = parse_servers(config)?;
288        let mode = parse_or_default::<Mode>(config, "mode")?;
289        let auth = parse_auth(config)?;
290        let tls = parse_tls(config)?;
291
292        let subject = match (config.get("subject"), config.get("subject.column")) {
293            (Some(s), None) => SubjectSpec::Literal(s.to_string()),
294            (None, Some(col)) => SubjectSpec::Column(col.to_string()),
295            (Some(_), Some(_)) => {
296                return Err(cfg_err(
297                    "[LDB-5050] set 'subject' OR 'subject.column', not both",
298                ));
299            }
300            (None, None) => {
301                return Err(cfg_err(
302                    "[LDB-5051] 'subject' or 'subject.column' is required",
303                ));
304            }
305        };
306
307        let dedup_id_column = config.get("dedup.id.column").map(str::to_string);
308
309        let cfg = Self {
310            servers,
311            mode,
312            auth,
313            tls,
314            stream: config
315                .get("stream")
316                .map(str::trim)
317                .filter(|stream| !stream.is_empty())
318                .map(str::to_string),
319            subject,
320            dedup_id_column,
321            min_duplicate_window: require_positive_duration(
322                config,
323                "min.duplicate.window.ms",
324                Duration::from_secs(120),
325            )?,
326            max_pending: require_positive_usize(config, "max.pending", MAX_PENDING_PUBLISH_ACKS)?,
327            ack_timeout: require_positive_duration(
328                config,
329                "ack.timeout.ms",
330                Duration::from_secs(30),
331            )?,
332            format: parse_format(config)?,
333            header_columns: config
334                .get("header.columns")
335                .map(split_csv)
336                .unwrap_or_default()
337                .into_iter()
338                .map(|name| {
339                    if !name
340                        .bytes()
341                        .all(|byte| (33..=126).contains(&byte) && byte != b':')
342                    {
343                        return Err(cfg_err(&format!(
344                            "[LDB-5066] header.columns entry '{name}' is not a valid ASCII NATS header name"
345                        )));
346                    }
347                    HeaderName::from_str(&name).map_err(|error| {
348                        cfg_err(&format!(
349                            "[LDB-5066] header.columns entry '{name}' is not a valid NATS header name: {error}"
350                        ))
351                    })
352                })
353                .collect::<Result<_, _>>()?,
354        };
355        cfg.validate()?;
356        Ok(cfg)
357    }
358
359    fn validate(&self) -> Result<(), ConnectorError> {
360        if self.max_pending > MAX_PENDING_PUBLISH_ACKS {
361            return Err(cfg_err(&format!(
362                "max.pending must be <= {MAX_PENDING_PUBLISH_ACKS}"
363            )));
364        }
365        if self.ack_timeout > MAX_NATS_IO_TIMEOUT {
366            return Err(cfg_err(&format!(
367                "ack.timeout.ms must be <= {}",
368                MAX_NATS_IO_TIMEOUT.as_millis()
369            )));
370        }
371        if self.mode == Mode::Core && self.dedup_id_column.is_some() {
372            return Err(cfg_err(
373                "[LDB-5053] 'dedup.id.column' requires mode=jetstream because core NATS has no \
374                 server-side Nats-Msg-Id deduplication",
375            ));
376        }
377        if self.dedup_id_column.is_some() && self.stream.is_none() {
378            return Err(cfg_err(
379                "[LDB-5055] 'dedup.id.column' requires 'stream' so the sink can validate its \
380                 duplicate_window at startup",
381            ));
382        }
383        for h in &self.header_columns {
384            if is_reserved_header(h.as_ref()) {
385                return Err(cfg_err(&format!(
386                    "[LDB-5065] header.columns entry '{h}' collides with a reserved NATS \
387                     header (Nats-Msg-Id / Nats-Expected-Stream); rename the column",
388                )));
389            }
390        }
391        match self.mode {
392            Mode::JetStream if self.stream.is_none() => {
393                return Err(cfg_err(
394                    "[LDB-5071] mode=jetstream requires an explicit 'stream' so durable storage, \
395                     replica count, and publish routing can be validated",
396                ));
397            }
398            Mode::Core if self.stream.is_some() => {
399                return Err(cfg_err(
400                    "[LDB-5071] 'stream' is valid only in mode=jetstream; remove it for core NATS",
401                ));
402            }
403            Mode::Core | Mode::JetStream => {}
404        }
405        Ok(())
406    }
407}
408
409/// Header names the sink manages itself; a user header with the same
410/// name would otherwise silently clobber bounded dedup or stream fencing.
411fn is_reserved_header(name: &str) -> bool {
412    const RESERVED: &[&str] = &["Nats-Msg-Id", "Nats-Expected-Stream"];
413    RESERVED.iter().any(|r| r.eq_ignore_ascii_case(name))
414}
415
416// ── helpers ──
417
418fn cfg_err(msg: &str) -> ConnectorError {
419    ConnectorError::ConfigurationError(msg.to_string())
420}
421
422fn parse_servers(config: &ConnectorConfig) -> Result<Vec<String>, ConnectorError> {
423    let raw = config.require("servers")?;
424    let servers: Vec<String> = split_csv(raw);
425    if servers.is_empty() {
426        return Err(cfg_err("[LDB-5030] 'servers' must not be empty"));
427    }
428    Ok(servers)
429}
430
431fn split_csv(raw: &str) -> Vec<String> {
432    raw.split(',')
433        .map(str::trim)
434        .filter(|s| !s.is_empty())
435        .map(str::to_string)
436        .collect()
437}
438
439fn parse_or_default<T: std::str::FromStr + Default>(
440    config: &ConnectorConfig,
441    key: &str,
442) -> Result<T, ConnectorError>
443where
444    T::Err: std::fmt::Display,
445{
446    match config.get(key) {
447        Some(s) => s
448            .parse()
449            .map_err(|e: T::Err| cfg_err(&format!("invalid {key}: {e}"))),
450        None => Ok(T::default()),
451    }
452}
453
454fn parse_format(config: &ConnectorConfig) -> Result<Format, ConnectorError> {
455    match config.get("format") {
456        Some(s) => Format::parse(s).map_err(|e| cfg_err(&e.to_string())),
457        None => Ok(Format::Json),
458    }
459}
460
461fn parse_bool(config: &ConnectorConfig, key: &str, default: bool) -> Result<bool, ConnectorError> {
462    match config.get(key) {
463        Some(s) => s
464            .parse::<bool>()
465            .map_err(|_| cfg_err(&format!("{key} must be 'true' or 'false', got '{s}'"))),
466        None => Ok(default),
467    }
468}
469
470fn parse_opt_u64(config: &ConnectorConfig, key: &str) -> Result<Option<u64>, ConnectorError> {
471    config.get(key).map(|s| parse_int(key, s)).transpose()
472}
473
474fn parse_i64(config: &ConnectorConfig, key: &str, default: i64) -> Result<i64, ConnectorError> {
475    config.get(key).map_or(Ok(default), |s| parse_int(key, s))
476}
477
478fn parse_usize(
479    config: &ConnectorConfig,
480    key: &str,
481    default: usize,
482) -> Result<usize, ConnectorError> {
483    config.get(key).map_or(Ok(default), |s| parse_int(key, s))
484}
485
486fn require_positive_duration(
487    config: &ConnectorConfig,
488    key: &str,
489    default: Duration,
490) -> Result<Duration, ConnectorError> {
491    let v = parse_duration_ms(config, key, default)?;
492    if v.is_zero() {
493        return Err(cfg_err(&format!("{key} must be > 0")));
494    }
495    Ok(v)
496}
497
498fn require_positive_usize(
499    config: &ConnectorConfig,
500    key: &str,
501    default: usize,
502) -> Result<usize, ConnectorError> {
503    let v = parse_usize(config, key, default)?;
504    if v == 0 {
505        return Err(cfg_err(&format!("{key} must be > 0")));
506    }
507    Ok(v)
508}
509
510/// NATS allows `-1` for "unlimited" in `max_deliver` / `max_ack_pending`.
511/// Reject 0 (undefined) and negative values other than -1.
512fn require_positive_or_unlimited_i64(
513    config: &ConnectorConfig,
514    key: &str,
515    default: i64,
516) -> Result<i64, ConnectorError> {
517    let v = parse_i64(config, key, default)?;
518    if v == 0 || v < -1 {
519        return Err(cfg_err(&format!("{key} must be > 0, or -1 for unlimited")));
520    }
521    Ok(v)
522}
523
524fn parse_duration_ms(
525    config: &ConnectorConfig,
526    key: &str,
527    default: Duration,
528) -> Result<Duration, ConnectorError> {
529    config.get(key).map_or(Ok(default), |s| {
530        parse_int::<u64>(key, s).map(Duration::from_millis)
531    })
532}
533
534fn parse_int<T: std::str::FromStr>(key: &str, raw: &str) -> Result<T, ConnectorError> {
535    raw.parse::<T>()
536        .map_err(|_| cfg_err(&format!("{key} must be an integer, got '{raw}'")))
537}
538
539fn parse_auth(config: &ConnectorConfig) -> Result<AuthMode, ConnectorError> {
540    let mode = config.get("auth.mode").unwrap_or("none");
541    match mode {
542        "none" | "" => {
543            if config.get("user").is_some()
544                || config.get("password").is_some()
545                || config.get("token").is_some()
546                || config.get("creds.file").is_some()
547            {
548                return Err(cfg_err(
549                    "[LDB-5063] credentials set but auth.mode=none; \
550                     set auth.mode explicitly or remove the credentials",
551                ));
552            }
553            Ok(AuthMode::None)
554        }
555        "user_pass" => {
556            let user = config
557                .get("user")
558                .ok_or_else(|| cfg_err("[LDB-5060] auth.mode=user_pass requires 'user'"))?
559                .to_string();
560            let password = config
561                .get("password")
562                .ok_or_else(|| cfg_err("[LDB-5060] auth.mode=user_pass requires 'password'"))?
563                .to_string();
564            Ok(AuthMode::UserPass { user, password })
565        }
566        "token" => {
567            let token = config
568                .get("token")
569                .ok_or_else(|| cfg_err("[LDB-5061] auth.mode=token requires 'token'"))?
570                .to_string();
571            Ok(AuthMode::Token(token))
572        }
573        "creds_file" => {
574            let path = config
575                .get("creds.file")
576                .ok_or_else(|| cfg_err("[LDB-5064] auth.mode=creds_file requires 'creds.file'"))?
577                .to_string();
578            Ok(AuthMode::CredsFile(path))
579        }
580        other => Err(cfg_err(&format!(
581            "invalid auth.mode '{other}'; expected none | user_pass | token | creds_file"
582        ))),
583    }
584}
585
586fn parse_tls(config: &ConnectorConfig) -> Result<TlsConfig, ConnectorError> {
587    let cert_location = config.get("tls.cert.location").map(str::to_string);
588    let key_location = config.get("tls.key.location").map(str::to_string);
589    if cert_location.is_some() != key_location.is_some() {
590        return Err(cfg_err(
591            "[LDB-5062] tls.cert.location and tls.key.location must both be set \
592             (mutual-TLS client cert) or both be unset",
593        ));
594    }
595    Ok(TlsConfig {
596        enabled: parse_bool(config, "tls.enabled", false)?,
597        ca_location: config.get("tls.ca.location").map(str::to_string),
598        cert_location,
599        key_location,
600    })
601}
602
603/// Build `ConnectOptions` from parsed auth + TLS. Shared between
604/// source and sink.
605///
606/// # Errors
607///
608/// Returns `ConnectorError` if the creds file can't be read or parsed.
609pub(super) fn build_connect_options(
610    auth: &AuthMode,
611    tls: &TlsConfig,
612) -> Result<async_nats::ConnectOptions, ConnectorError> {
613    let mut opts = async_nats::ConnectOptions::new();
614    match auth {
615        AuthMode::None => {}
616        AuthMode::UserPass { user, password } => {
617            opts = opts.user_and_password(user.clone(), password.clone());
618        }
619        AuthMode::Token(token) => {
620            opts = opts.token(token.clone());
621        }
622        AuthMode::CredsFile(path) => {
623            let contents = std::fs::read_to_string(path)
624                .map_err(|e| cfg_err(&format!("creds.file '{path}': {e}")))?;
625            opts = async_nats::ConnectOptions::with_credentials(&contents)
626                .map_err(|e| cfg_err(&format!("creds.file '{path}' invalid: {e}")))?;
627        }
628    }
629    let tls_touched = tls.enabled || tls.ca_location.is_some() || tls.cert_location.is_some();
630    if tls_touched {
631        opts = opts.require_tls(true);
632    }
633    if let Some(ca) = &tls.ca_location {
634        opts = opts.add_root_certificates(ca.into());
635    }
636    if let (Some(cert), Some(key)) = (&tls.cert_location, &tls.key_location) {
637        opts = opts.add_client_certificate(cert.into(), key.into());
638    }
639    Ok(opts
640        .connection_timeout(NATS_CONNECTION_TIMEOUT)
641        .request_timeout(Some(NATS_REQUEST_TIMEOUT)))
642}
643
644#[cfg(test)]
645#[allow(clippy::disallowed_types)]
646mod tests;
laminar_connectors/nats/config/mod.rs

laminar_connectors/nats/config/
mod.rs
