rocketmq_client_rust/nameserver_discovery/
config.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt;
16use std::num::NonZeroU16;
17use std::sync::Arc;
18use std::time::Duration;
19
20use crate::ClientError;
21use crate::ClientResult;
22
23use crate::config_support::name_server_target::parse_legacy_namesrv_addr;
24
25const DEFAULT_MIN_REFRESH: Duration = Duration::from_secs(5);
26const DEFAULT_MAX_REFRESH: Duration = Duration::from_secs(60);
27const DEFAULT_STALE_MAX: Duration = Duration::from_secs(5 * 60);
28const DEFAULT_DRAIN_TIMEOUT: Duration = Duration::from_secs(30);
29const DEFAULT_ENDPOINT_LIMIT: usize = 64;
30const MAX_ENDPOINT_LIMIT: usize = 64;
31
32/// A normalized ASCII DNS name used as a logical NameServer authority.
33#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
34pub struct DnsName(Arc<str>);
35
36impl DnsName {
37    /// Parses and normalizes a DNS name.
38    pub fn parse(value: impl AsRef<str>) -> ClientResult<Self> {
39        let normalized = value.as_ref().trim().trim_end_matches('.').to_ascii_lowercase();
40        validate_dns_name(&normalized)?;
41        Ok(Self(Arc::from(normalized)))
42    }
43
44    /// Returns the canonical DNS name without a trailing dot.
45    #[must_use]
46    pub fn as_str(&self) -> &str {
47        &self.0
48    }
49}
50
51impl fmt::Display for DnsName {
52    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53        formatter.write_str(self.as_str())
54    }
55}
56
57/// A canonical `host:port` NameServer authority.
58#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
59pub struct NameServerAuthority(Arc<str>);
60
61impl NameServerAuthority {
62    /// Parses exactly one canonical NameServer authority.
63    pub fn parse(value: impl AsRef<str>) -> ClientResult<Self> {
64        let parsed = parse_legacy_namesrv_addr(value.as_ref())?;
65        let addresses = parsed.into_addresses();
66        if addresses.len() != 1 {
67            return Err(invalid_config(
68                "nameserver_discovery.static",
69                value.as_ref(),
70                "must contain exactly one host:port authority",
71            ));
72        }
73        Ok(Self(Arc::from(addresses[0].as_str())))
74    }
75
76    /// Returns the canonical authority.
77    #[must_use]
78    pub fn as_str(&self) -> &str {
79        &self.0
80    }
81}
82
83impl fmt::Display for NameServerAuthority {
84    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
85        formatter.write_str(self.as_str())
86    }
87}
88
89/// Supported NameServer discovery sources.
90#[non_exhaustive]
91#[derive(Clone, Debug, Eq, PartialEq)]
92pub enum NameServerSource {
93    /// A fixed canonical list. This preserves existing transport behavior.
94    Static(Vec<NameServerAuthority>),
95    /// DNS A/AAAA discovery for a logical authority.
96    Dns { host: DnsName, port: NonZeroU16 },
97}
98
99impl NameServerSource {
100    /// Creates a DNS A/AAAA source.
101    pub fn dns(host: impl AsRef<str>, port: u16) -> ClientResult<Self> {
102        let port = NonZeroU16::new(port)
103            .ok_or_else(|| invalid_config("nameserver_discovery.port", port, "must be between 1 and 65535"))?;
104        Ok(Self::Dns {
105            host: DnsName::parse(host)?,
106            port,
107        })
108    }
109
110    /// Creates a validated static source.
111    pub fn static_endpoints(endpoints: Vec<NameServerAuthority>) -> ClientResult<Self> {
112        if endpoints.is_empty() {
113            return Err(invalid_config(
114                "nameserver_discovery.static",
115                "[]",
116                "must contain at least one endpoint",
117            ));
118        }
119        if endpoints.len() > MAX_ENDPOINT_LIMIT {
120            return Err(invalid_config(
121                "nameserver_discovery.static",
122                endpoints.len(),
123                "must not contain more than 64 endpoints",
124            ));
125        }
126        Ok(Self::Static(endpoints))
127    }
128}
129
130/// Typed NameServer discovery configuration.
131#[derive(Clone, Debug, Eq, PartialEq)]
132pub struct NameServerDiscoveryConfig {
133    source: NameServerSource,
134    min_refresh: Duration,
135    max_refresh: Duration,
136    stale_max: Duration,
137    drain_timeout: Duration,
138    endpoint_limit: usize,
139}
140
141impl NameServerDiscoveryConfig {
142    /// Creates a discovery configuration with cloud-native defaults.
143    #[must_use]
144    pub fn new(source: NameServerSource) -> Self {
145        Self {
146            source,
147            min_refresh: DEFAULT_MIN_REFRESH,
148            max_refresh: DEFAULT_MAX_REFRESH,
149            stale_max: DEFAULT_STALE_MAX,
150            drain_timeout: DEFAULT_DRAIN_TIMEOUT,
151            endpoint_limit: DEFAULT_ENDPOINT_LIMIT,
152        }
153    }
154
155    /// Sets the inclusive refresh interval clamp.
156    pub fn with_refresh_bounds(mut self, min: Duration, max: Duration) -> ClientResult<Self> {
157        if min.is_zero() || max < min {
158            return Err(invalid_config(
159                "nameserver_discovery.refresh",
160                format!("{min:?}..={max:?}"),
161                "minimum must be non-zero and no greater than maximum",
162            ));
163        }
164        self.min_refresh = min;
165        self.max_refresh = max;
166        Ok(self)
167    }
168
169    /// Sets how long the last-known-good endpoint set may remain usable.
170    pub fn with_stale_max(mut self, stale_max: Duration) -> ClientResult<Self> {
171        if stale_max.is_zero() {
172            return Err(invalid_config(
173                "nameserver_discovery.stale_max",
174                "0",
175                "must be non-zero",
176            ));
177        }
178        self.stale_max = stale_max;
179        Ok(self)
180    }
181
182    /// Sets how long a removed endpoint may finish already in-flight work.
183    pub fn with_drain_timeout(mut self, drain_timeout: Duration) -> ClientResult<Self> {
184        if drain_timeout.is_zero() {
185            return Err(invalid_config(
186                "nameserver_discovery.drain_timeout",
187                "0",
188                "must be non-zero",
189            ));
190        }
191        self.drain_timeout = drain_timeout;
192        Ok(self)
193    }
194
195    /// Sets the resolved endpoint cap. Values above 64 are rejected.
196    pub fn with_endpoint_limit(mut self, endpoint_limit: usize) -> ClientResult<Self> {
197        if !(1..=MAX_ENDPOINT_LIMIT).contains(&endpoint_limit) {
198            return Err(invalid_config(
199                "nameserver_discovery.endpoint_limit",
200                endpoint_limit,
201                "must be between 1 and 64",
202            ));
203        }
204        self.endpoint_limit = endpoint_limit;
205        Ok(self)
206    }
207
208    /// Returns the configured source.
209    #[must_use]
210    pub fn source(&self) -> &NameServerSource {
211        &self.source
212    }
213
214    #[must_use]
215    pub(crate) fn min_refresh(&self) -> Duration {
216        self.min_refresh
217    }
218
219    #[must_use]
220    pub(crate) fn max_refresh(&self) -> Duration {
221        self.max_refresh
222    }
223
224    #[must_use]
225    pub(crate) fn stale_max(&self) -> Duration {
226        self.stale_max
227    }
228
229    #[must_use]
230    pub(crate) fn drain_timeout(&self) -> Duration {
231        self.drain_timeout
232    }
233
234    #[must_use]
235    pub(crate) fn endpoint_limit(&self) -> usize {
236        self.endpoint_limit
237    }
238
239    pub(crate) fn validate(&self) -> ClientResult<()> {
240        match &self.source {
241            NameServerSource::Static(endpoints) if endpoints.is_empty() => Err(invalid_config(
242                "nameserver_discovery.static",
243                "[]",
244                "must contain at least one endpoint",
245            )),
246            NameServerSource::Static(endpoints) if endpoints.len() > self.endpoint_limit => Err(invalid_config(
247                "nameserver_discovery.static",
248                endpoints.len(),
249                "exceeds the configured endpoint limit",
250            )),
251            _ => Ok(()),
252        }
253    }
254
255    pub(crate) fn fingerprint(&self) -> String {
256        let source = match &self.source {
257            NameServerSource::Static(endpoints) => {
258                let mut endpoints = endpoints.iter().map(NameServerAuthority::as_str).collect::<Vec<_>>();
259                endpoints.sort_unstable();
260                endpoints.dedup();
261                format!("static:{}", endpoints.join(";"))
262            }
263            NameServerSource::Dns { host, port } => format!("dns:{}:{}", host.as_str(), port.get()),
264        };
265        format!(
266            "{source}|refresh={}-{}|stale={}|drain={}|limit={}",
267            self.min_refresh.as_millis(),
268            self.max_refresh.as_millis(),
269            self.stale_max.as_millis(),
270            self.drain_timeout.as_millis(),
271            self.endpoint_limit
272        )
273    }
274
275    pub(crate) fn static_canonical(&self) -> Option<String> {
276        let NameServerSource::Static(endpoints) = &self.source else {
277            return None;
278        };
279        let mut seen = std::collections::HashSet::new();
280        Some(
281            endpoints
282                .iter()
283                .filter(|endpoint| seen.insert(endpoint.as_str()))
284                .map(NameServerAuthority::as_str)
285                .collect::<Vec<_>>()
286                .join(";"),
287        )
288    }
289}
290
291fn validate_dns_name(value: &str) -> ClientResult<()> {
292    if value.is_empty() || value.len() > 253 {
293        return Err(invalid_config(
294            "nameserver_discovery.dns_name",
295            value,
296            "must contain between 1 and 253 ASCII characters",
297        ));
298    }
299    for label in value.split('.') {
300        if label.is_empty() || label.len() > 63 {
301            return Err(invalid_config(
302                "nameserver_discovery.dns_name",
303                value,
304                "each DNS label must contain between 1 and 63 characters",
305            ));
306        }
307        if !label
308            .bytes()
309            .all(|character| character.is_ascii_alphanumeric() || matches!(character, b'-' | b'_'))
310            || label.starts_with('-')
311            || label.ends_with('-')
312        {
313            return Err(invalid_config(
314                "nameserver_discovery.dns_name",
315                value,
316                "contains an invalid DNS label",
317            ));
318        }
319    }
320    Ok(())
321}
322
323fn invalid_config(key: &'static str, value: impl ToString, reason: impl Into<String>) -> ClientError {
324    ClientError::config_invalid(key, value.to_string(), reason.into())
325}
326
327#[cfg(test)]
328mod tests {
329    use super::*;
330
331    #[test]
332    fn dns_source_normalizes_name_and_rejects_invalid_port() {
333        let source = NameServerSource::dns("NameSrv.Default.Svc.", 9876).expect("DNS source");
334        assert!(matches!(
335            source,
336            NameServerSource::Dns { ref host, port }
337                if host.as_str() == "namesrv.default.svc" && port.get() == 9876
338        ));
339        assert!(NameServerSource::dns("namesrv.default.svc", 0).is_err());
340    }
341
342    #[test]
343    fn discovery_fingerprint_is_stable_for_equivalent_static_sets() {
344        let first = NameServerDiscoveryConfig::new(
345            NameServerSource::static_endpoints(vec![
346                NameServerAuthority::parse("ns-b:9876").unwrap(),
347                NameServerAuthority::parse("ns-a:9876").unwrap(),
348            ])
349            .unwrap(),
350        );
351        let second = NameServerDiscoveryConfig::new(
352            NameServerSource::static_endpoints(vec![
353                NameServerAuthority::parse("NS-A:9876").unwrap(),
354                NameServerAuthority::parse("ns-b:9876").unwrap(),
355            ])
356            .unwrap(),
357        );
358        assert_eq!(first.fingerprint(), second.fingerprint());
359    }
360}
rocketmq_client_rust/nameserver_discovery/config.rs

rocketmq_client_rust/nameserver_discovery/
config.rs
