Skip to main content

nadisc_runtime/
assembly.rs

1//! 多后端类型化装配。输入已经完成占位符解析,公共组件不依赖任何应用配置框架。
2
3use crate::{
4    config::{invalid, validate_name, ConnectionPolicy, Limits, RegistrationPolicy},
5    providers::LocalProvider,
6    ConnectionSpec, DiscoveryOwner, Eligibility, OwnerBuilder, RegistrationSpec,
7};
8use nadisc::{
9    managed::{ManagedProvider, RegistrationPlan, Result},
10    DnsService, Instance,
11};
12use serde::{
13    de::{Error, MapAccess, Visitor},
14    Deserialize, Serialize,
15};
16use std::{
17    collections::{BTreeMap, HashMap},
18    sync::Arc,
19};
20
21/// 多注册中心连接、发布目标与资源预算;配置冻结后由同一 owner 管理生命周期。
22#[derive(Clone, Default, Deserialize, Serialize)]
23#[serde(default, deny_unknown_fields)]
24pub struct DiscoveryConfig {
25    /// 是否启用发现;禁用时不得同时声明连接、注册或资格探针。
26    pub enabled: bool,
27    /// 是否启用宿主本地资格探针;不以远端注册结果反向驱动本地资格。
28    pub eligibility_probe: bool,
29    /// 兼容显式默认引用;省略时只采用名为 primary 的连接,不按数量或顺序推断。
30    #[serde(skip_serializing_if = "Option::is_none")]
31    pub default_connection: Option<String>,
32    /// 所有连接和发布目标共享的有限资源预算。
33    pub limits: Limits,
34    /// 以 qualifier 为键的连接目录,不按顺序推断默认连接。
35    #[serde(deserialize_with = "unique_map")]
36    pub connections: BTreeMap<String, ConnectionConfig>,
37    /// 注册目标集合,包含各目标独立的状态或发布策略。
38    #[serde(deserialize_with = "unique_map")]
39    pub registrations: BTreeMap<String, RegistrationConfig>,
40}
41
42/// 单个 qualifier 的后端配置;provider 必须与唯一的后端配置块一致。
43#[derive(Clone, Deserialize, Serialize)]
44#[serde(default, deny_unknown_fields)]
45pub struct ConnectionConfig {
46    /// 后端类型或实例,决定该连接支持的发现及注册能力。
47    pub provider: String,
48    /// 是否要求启动阶段确认成功;失败时阻止宿主完成启动。
49    pub startup_required: bool,
50    /// 一次请求的最大耗时,单位为毫秒。
51    pub request_timeout_ms: u64,
52    /// 后台轮询间隔,单位为毫秒,必须留出请求和证据时效预算。
53    pub poll_interval_ms: u64,
54    /// 无人使用的服务观察保留时长,单位为毫秒。
55    pub idle_timeout_ms: u64,
56    /// 共享发现证据的年龄上限与异常宽限。
57    pub snapshot: SnapshotPolicy,
58    /// 连接准备失败时的重试退避预算。
59    pub retry: RetryPolicy,
60    /// Nacos 专用配置,只有 Nacos 后端可以使用。
61    #[serde(deserialize_with = "optional_unique_value")]
62    pub nacos: Option<serde_json::Value>,
63    /// Consul 专用配置,只有 Consul 后端可以使用。
64    #[serde(deserialize_with = "optional_unique_value")]
65    pub consul: Option<serde_json::Value>,
66    /// Kubernetes 只读发现配置,不授予 Service 或 EndpointSlice 写入权。
67    #[serde(deserialize_with = "optional_unique_value")]
68    pub kubernetes: Option<serde_json::Value>,
69    /// 静态实例映射;配置文件中的键为 static。
70    #[serde(rename = "static")]
71    pub static_config: Option<StaticConfig>,
72    /// DNS 服务映射,不支持服务注册。
73    pub dns: Option<DnsConfig>,
74    /// 该连接的 REST 调用策略,与其他 qualifier 的调用预算隔离。
75    pub rest: Option<RestPolicy>,
76}
77
78/// 宿主 REST 桥的独立调用策略;共享发现时效仅由 snapshot 策略裁决。
79#[derive(Clone, Deserialize, Serialize)]
80#[serde(default, deny_unknown_fields)]
81pub struct RestPolicy {
82    /// 一次请求的最大耗时,单位为毫秒。
83    pub request_timeout_ms: u64,
84    /// HTTP 建连预算,单位为毫秒,不得超过请求总预算。
85    pub connect_timeout_ms: u64,
86    /// 实例选择策略,可使用 round_robin 或 weighted。
87    pub lb_strategy: String,
88    /// 发现实例的 HTTP 协议,只接受 http 或 https。
89    pub instance_scheme: String,
90    /// 是否允许 GET/HEAD 在传输错误后按剩余预算重试。
91    pub retry_get_head_on_transport_error: bool,
92    /// 是否允许 GET/HEAD 在可重试响应状态后重试。
93    pub retry_get_head_on_retryable_status: bool,
94    /// 一次逻辑请求允许的总尝试次数,包含首次调用。
95    pub max_attempts: usize,
96}
97impl Default for RestPolicy {
98    /// 业务作用:给各连接使用独立有限的 HTTP 调用预算,默认不重试。
99    /// 参数说明:无。
100    /// 返回:十秒总预算、两秒连接预算和轮询分配策略。
101    fn default() -> Self {
102        Self {
103            request_timeout_ms: 10000,
104            connect_timeout_ms: 2000,
105            lb_strategy: "round_robin".into(),
106            instance_scheme: "http".into(),
107            retry_get_head_on_transport_error: false,
108            retry_get_head_on_retryable_status: false,
109            max_attempts: 1,
110        }
111    }
112}
113impl RestPolicy {
114    /// 业务作用:在连接准备前拒绝非法调用时长、协议、分配策略和重试次数。
115    /// 参数说明:无。
116    /// 返回:策略有效时成功;不开放任意 HTTP 方法的自动重试。
117    pub fn validate(&self) -> Result<()> {
118        if self.request_timeout_ms == 0
119            || self.request_timeout_ms > 86_400_000
120            || self.connect_timeout_ms == 0
121            || self.connect_timeout_ms > self.request_timeout_ms
122            || !matches!(self.lb_strategy.as_str(), "round_robin" | "weighted")
123            || !matches!(self.instance_scheme.as_str(), "http" | "https")
124            || !(1..=16).contains(&self.max_attempts)
125        {
126            return Err(invalid("invalid per-connection REST policy"));
127        }
128        Ok(())
129    }
130}
131
132/// 发现证据的有效期与查询异常后的有限宽限,单位为毫秒。
133#[derive(Clone, Deserialize, Serialize)]
134#[serde(default, deny_unknown_fields)]
135pub struct SnapshotPolicy {
136    /// 发现证据的最大年龄,单位为毫秒。
137    pub max_age_ms: u64,
138    /// 查询异常后的旧证据宽限,单位为毫秒;零表示禁止异常复用。
139    pub stale_if_error_ms: u64,
140}
141impl Default for SnapshotPolicy {
142    /// 业务作用:默认禁止异常后继续使用旧路由。
143    /// 参数说明:无。
144    /// 返回:十五秒最大证据年龄和零异常宽限。
145    fn default() -> Self {
146        Self {
147            max_age_ms: 15000,
148            stale_if_error_ms: 0,
149        }
150    }
151}
152
153/// 连接准备失败后的有限退避策略,单位为毫秒。
154#[derive(Clone, Deserialize, Serialize)]
155#[serde(default, deny_unknown_fields)]
156pub struct RetryPolicy {
157    /// 连接恢复的初始退避,单位为毫秒。
158    pub initial_backoff_ms: u64,
159    /// 连接恢复退避的上限,单位为毫秒。
160    pub max_backoff_ms: u64,
161}
162impl Default for RetryPolicy {
163    /// 业务作用:对不可达连接限制恢复重试频率。
164    /// 参数说明:无。
165    /// 返回:一秒起步、三十秒封顶的退避。
166    fn default() -> Self {
167        Self {
168            initial_backoff_ms: 1000,
169            max_backoff_ms: 30000,
170        }
171    }
172}
173
174impl Default for ConnectionConfig {
175    /// 业务作用:默认连接必须准备成功,但没有指定 provider 时不猜测后端。
176    /// 参数说明:无。
177    /// 返回:有限请求和观察预算的候选配置。
178    fn default() -> Self {
179        Self {
180            provider: String::new(),
181            startup_required: true,
182            request_timeout_ms: 3000,
183            poll_interval_ms: 5000,
184            idle_timeout_ms: 30000,
185            snapshot: SnapshotPolicy::default(),
186            retry: RetryPolicy::default(),
187            nacos: None,
188            consul: None,
189            kubernetes: None,
190            static_config: None,
191            dns: None,
192            rest: None,
193        }
194    }
195}
196
197/// 由配置提供的固定实例集合,不向注册中心写入。
198#[derive(Clone, Deserialize, Serialize)]
199#[serde(deny_unknown_fields)]
200pub struct StaticConfig {
201    /// 按逻辑服务名绑定的目标,不接受重复键。
202    #[serde(deserialize_with = "unique_map")]
203    pub services: BTreeMap<String, Vec<Instance>>,
204}
205/// 逻辑服务到 DNS 名称和端口的显式映射。
206#[derive(Clone, Deserialize, Serialize)]
207#[serde(deny_unknown_fields)]
208pub struct DnsConfig {
209    /// 按逻辑服务名绑定的目标,不接受重复键。
210    #[serde(deserialize_with = "unique_map")]
211    pub services: BTreeMap<String, DnsTarget>,
212}
213/// DNS 查询目标;端口由配置确定,不从名称推断。
214#[derive(Clone, Deserialize, Serialize)]
215#[serde(deny_unknown_fields)]
216pub struct DnsTarget {
217    /// 需要解析的 DNS 主机名。
218    pub host: String,
219    /// 目标监听端口;注册配置省略时由 listener_ref 解析。
220    pub port: u16,
221}
222
223/// 单个服务发布目标及其启动、健康回查和摘流策略。
224#[derive(Clone, Deserialize, Serialize)]
225#[serde(default, deny_unknown_fields)]
226pub struct RegistrationConfig {
227    /// 发布目标的 qualifier;Rust 字段保留兼容名称,配置序列化使用 qualifier。
228    #[serde(rename = "qualifier", alias = "connection_ref")]
229    pub connection_ref: String,
230    /// 发布到目标后端的服务名称,必须存在对应服务绑定。
231    pub service_name: String,
232    /// 对外可达地址;注册配置可显式覆盖宿主监听地址。
233    pub ip: Option<String>,
234    /// 目标监听端口;注册配置省略时由 listener_ref 解析。
235    pub port: Option<u16>,
236    /// 宿主已绑定监听器的引用,用于解析未显式指定的地址与端口。
237    pub listener_ref: Option<String>,
238    /// 宿主本地业务资格引用,控制是否允许发布实例。
239    pub eligibility_ref: Option<String>,
240    /// 随注册写入的实例属性,不得携带凭据。
241    pub metadata: HashMap<String, String>,
242    /// 是否要求启动阶段确认成功;失败时阻止宿主完成启动。
243    pub startup_required: bool,
244    /// 远端注册失去有效健康证据时是否影响宿主整体就绪。
245    pub readiness_critical: bool,
246    /// 注册健康回查间隔,单位为毫秒。
247    pub monitor_interval_ms: u64,
248    /// 注册健康证据的有效窗口,单位为毫秒。
249    pub observation_grace_ms: u64,
250    /// 单项注销预算,单位为毫秒,仍受宿主关闭总期限限制。
251    pub deregister_timeout_ms: u64,
252    /// Nacos 专用配置,只有 Nacos 后端可以使用。
253    pub nacos: Option<NacosRegistration>,
254    /// Consul 专用配置,只有 Consul 后端可以使用。
255    #[serde(deserialize_with = "optional_unique_value")]
256    pub consul: Option<serde_json::Value>,
257}
258
259impl Default for RegistrationConfig {
260    /// 业务作用:注册目标省略 qualifier 时使用 primary,默认要求首次注册确认。
261    /// 参数说明:无。
262    /// 返回:服务与地址待显式配置或宿主解析的注册项;兼容默认连接覆盖不改变此处的 primary。
263    fn default() -> Self {
264        Self {
265            connection_ref: "primary".into(),
266            service_name: String::new(),
267            ip: None,
268            port: None,
269            listener_ref: None,
270            eligibility_ref: None,
271            metadata: HashMap::new(),
272            startup_required: true,
273            readiness_critical: false,
274            monitor_interval_ms: 5000,
275            observation_grace_ms: 15000,
276            deregister_timeout_ms: 3000,
277            nacos: None,
278            consul: None,
279        }
280    }
281}
282
283/// Nacos 注册专用属性,不用于其他后端。
284#[derive(Clone, Deserialize, Serialize)]
285#[serde(default, deny_unknown_fields)]
286pub struct NacosRegistration {
287    /// 是否发布 Nacos 临时实例,由客户端维持其存活。
288    pub ephemeral: bool,
289    /// Nacos 实例所属集群名称。
290    pub cluster_name: String,
291    /// Nacos 实例选择权重。
292    pub weight: f64,
293}
294impl Default for NacosRegistration {
295    /// 业务作用:显式提供 Nacos 临时实例和默认集群属性。
296    /// 参数说明:无。
297    /// 返回:默认权重为一,不影响其他后端。
298    fn default() -> Self {
299        Self {
300            ephemeral: true,
301            cluster_name: "DEFAULT".into(),
302            weight: 1.0,
303        }
304    }
305}
306
307/// 宿主已绑定 listener 的事实;代理映射地址由 registration 的 ip/port 显式覆盖。
308#[derive(Clone)]
309pub struct ListenerEndpoint {
310    /// 对外可达地址;注册配置可显式覆盖宿主监听地址。
311    pub ip: String,
312    /// 目标监听端口;注册配置省略时由 listener_ref 解析。
313    pub port: u16,
314}
315
316impl DiscoveryConfig {
317    /// 业务作用:校验所有引用与 provider 块,必需注册提升所属连接的启动要求。
318    /// 参数说明:无。
319    /// 返回:配置可用于独立宿主或 napp;禁用配置不产生后端连接,引用与 provider 拒绝携带字段路径。
320    pub fn validate(&self) -> Result<()> {
321        self.limits
322            .validate()
323            .map_err(|error| error.at_config_path("discovery.limits"))?;
324        if !self.enabled
325            && (!self.connections.is_empty()
326                || !self.registrations.is_empty()
327                || self.default_connection.is_some()
328                || self.eligibility_probe)
329        {
330            return Err(invalid(
331                "disabled discovery cannot declare connections, registrations or probes",
332            ));
333        }
334        if self.connections.len() > self.limits.max_connections
335            || self.registrations.len() > self.limits.max_registrations
336        {
337            return Err(invalid("discovery configuration exceeds capacity"));
338        }
339        if let Some(name) = &self.default_connection {
340            if !self.connections.contains_key(name) {
341                return Err(invalid("unknown default connection")
342                    .at_config_path("discovery.default_connection"));
343            }
344        }
345        for (name, connection) in &self.connections {
346            validate_name(name).map_err(|error| error.at_config_path("discovery.connections"))?;
347            connection
348                .validate()
349                .map_err(|error| error.at_config_path(format!("discovery.connections.{name}")))?;
350        }
351        for (name, registration) in &self.registrations {
352            validate_name(name).map_err(|error| error.at_config_path("discovery.registrations"))?;
353            let connection = self
354                .connections
355                .get(&registration.connection_ref)
356                .ok_or_else(|| {
357                    invalid("unknown registration connection")
358                        .at_config_path(format!("discovery.registrations.{name}.qualifier"))
359                })?;
360            if !connection.has_service(&registration.service_name) {
361                return Err(invalid("unknown registration service binding")
362                    .at_config_path(format!("discovery.registrations.{name}.service_name")));
363            }
364            if registration.service_name.is_empty()
365                || registration.service_name.len() > 512
366                || registration
367                    .ip
368                    .as_ref()
369                    .is_some_and(|ip| ip.is_empty() || ip.trim() != ip)
370                || registration.port == Some(0)
371                || registration.metadata.len() > 128
372                || registration
373                    .metadata
374                    .iter()
375                    .any(|(key, value)| key.len() > 256 || value.len() > 4096)
376            {
377                return Err(invalid(
378                    "invalid registration service, endpoint or metadata",
379                ));
380            }
381            registration.policy().validate(&connection.policy())?;
382            match connection.provider.as_str() {
383                "nacos" if registration.consul.is_none() => {}
384                "consul" if registration.nacos.is_none() && registration.consul.is_some() => {}
385                _ => {
386                    return Err(invalid(
387                        "registration attributes do not match a writable provider",
388                    ))
389                }
390            }
391        }
392        Ok(())
393    }
394
395    /// 业务作用:构造各身份隔离的 provider 和公共 owner,不启动注册或网络轮询。
396    /// 参数说明:无;配置值应已完成宿主占位符解析。
397    /// 返回:唯一关闭 owner;只启用已编入的 provider。
398    pub async fn build(&self) -> Result<DiscoveryOwner> {
399        self.validate()?;
400        let mut builder = OwnerBuilder::default().limits(self.limits.clone());
401        if !self.enabled {
402            return builder.build();
403        }
404        if let Some(default) = &self.default_connection {
405            builder = builder.default_connection(default);
406        }
407        for (name, config) in &self.connections {
408            let mut policy = config.policy();
409            policy.startup_required |= self.registrations.values().any(|registration| {
410                registration.connection_ref == *name && registration.startup_required
411            });
412            builder = builder.connection(ConnectionSpec {
413                name: name.clone(),
414                provider: config.build().await.map_err(|error| {
415                    error.at_config_path(format!("discovery.connections.{name}"))
416                })?,
417                policy,
418            })?;
419        }
420        builder.build()
421    }
422
423    /// 业务作用:在全部监听器已绑定后解析一次完整注册集合。
424    /// 参数说明:`listeners` 是实际端点,`eligibility` 是宿主本地资格源;单项时允许省略对应引用。
425    /// 返回:全部可解析的计划;不执行外部注册,不生成业务专用 metadata。
426    pub fn resolve(
427        &self,
428        listeners: &BTreeMap<String, ListenerEndpoint>,
429        eligibility: &BTreeMap<String, Eligibility>,
430    ) -> Result<Vec<RegistrationSpec>> {
431        if !self.enabled {
432            return Ok(Vec::new());
433        }
434        self.validate()?;
435        let mut plans = Vec::new();
436        for (name, registration) in &self.registrations {
437            let listener = match &registration.listener_ref {
438                Some(name) => Some(
439                    listeners
440                        .get(name)
441                        .ok_or_else(|| invalid("unknown registration listener"))?,
442                ),
443                None if listeners.len() == 1 => listeners.values().next(),
444                None => None,
445            };
446            let inferred_listener = registration
447                .listener_ref
448                .as_ref()
449                .or_else(|| (listeners.len() == 1).then(|| listeners.keys().next().unwrap()));
450            let eligibility = match &registration.eligibility_ref {
451                Some(name) => eligibility.get(name),
452                None if inferred_listener.is_some() => eligibility.get(inferred_listener.unwrap()),
453                None if eligibility.len() == 1 => eligibility.values().next(),
454                None => None,
455            }
456            .cloned()
457            .ok_or_else(|| invalid("registration eligibility source is absent or ambiguous"))?;
458            let ip = registration
459                .ip
460                .as_ref()
461                .or_else(|| listener.map(|listener| &listener.ip))
462                .ok_or_else(|| invalid("registration IP is absent"))?;
463            let port = registration
464                .port
465                .or_else(|| listener.map(|listener| listener.port))
466                .ok_or_else(|| invalid("registration port is absent"))?;
467            let mut instance =
468                Instance::new(ip.clone(), port).with_metadata_map(registration.metadata.clone());
469            if let Some(nacos) = &registration.nacos {
470                instance = instance
471                    .with_cluster(&nacos.cluster_name)
472                    .with_ephemeral(nacos.ephemeral)
473                    .with_weight(nacos.weight);
474            }
475            #[allow(unused_mut)]
476            let mut plan = RegistrationPlan {
477                id: name.clone(),
478                service: registration.service_name.clone(),
479                instance,
480                attributes: BTreeMap::new(),
481            };
482            if let Some(consul) = &registration.consul {
483                #[cfg(feature = "consul")]
484                {
485                    #[derive(Deserialize)]
486                    #[serde(deny_unknown_fields)]
487                    struct CheckConfig {
488                        check: nadisc_consul::HealthCheck,
489                    }
490                    let config: CheckConfig = serde_json::from_value(consul.clone())
491                        .map_err(|_| invalid("invalid Consul registration configuration"))?;
492                    config.check.apply(&mut plan)?;
493                }
494                #[cfg(not(feature = "consul"))]
495                {
496                    let _ = consul;
497                    return Err(invalid("Consul provider feature is disabled"));
498                }
499            }
500            plans.push(RegistrationSpec {
501                connection_ref: registration.connection_ref.clone(),
502                plan,
503                policy: registration.policy(),
504                eligibility,
505            });
506        }
507        Ok(plans)
508    }
509}
510
511impl RegistrationConfig {
512    /// 业务作用:将配置中的启动与运行期要求转换成公共注册策略。
513    /// 参数说明:无。
514    /// 返回:两个必需性维度保持独立。
515    pub fn policy(&self) -> RegistrationPolicy {
516        RegistrationPolicy {
517            startup_required: self.startup_required,
518            readiness_critical: self.readiness_critical,
519            monitor_interval_ms: self.monitor_interval_ms,
520            observation_grace_ms: self.observation_grace_ms,
521            deregister_timeout_ms: self.deregister_timeout_ms,
522        }
523    }
524}
525
526impl ConnectionConfig {
527    /// 业务作用:在网络准备之前核对宿主引用是否属于该连接的显式服务目录。
528    /// 参数说明:`service` 是本地服务别名,不能替代后端服务名或跨连接查询。
529    /// 返回:静态、DNS 或已配置 provider 服务表包含该精确别名时为 true。
530    pub fn has_service(&self, service: &str) -> bool {
531        match self.provider.as_str() {
532            "static" => self
533                .static_config
534                .as_ref()
535                .is_some_and(|config| config.services.contains_key(service)),
536            "dns" => self
537                .dns
538                .as_ref()
539                .is_some_and(|config| config.services.contains_key(service)),
540            "nacos" | "consul" | "kubernetes" => {
541                let value = match self.provider.as_str() {
542                    "nacos" => &self.nacos,
543                    "consul" => &self.consul,
544                    _ => &self.kubernetes,
545                };
546                value
547                    .as_ref()
548                    .and_then(|value| value.get("services"))
549                    .and_then(serde_json::Value::as_object)
550                    .is_some_and(|services| services.contains_key(service))
551            }
552            _ => false,
553        }
554    }
555
556    /// 业务作用:把公开配置转换为 owner 的冻结时效预算。
557    /// 参数说明:无。
558    /// 返回:不依赖具体传输默认值的公共策略。
559    pub fn policy(&self) -> ConnectionPolicy {
560        ConnectionPolicy {
561            startup_required: self.startup_required,
562            request_timeout_ms: self.request_timeout_ms,
563            poll_interval_ms: self.poll_interval_ms,
564            max_age_ms: self.snapshot.max_age_ms,
565            stale_if_error_ms: self.snapshot.stale_if_error_ms,
566            idle_timeout_ms: self.idle_timeout_ms,
567            initial_backoff_ms: self.retry.initial_backoff_ms,
568            max_backoff_ms: self.retry.max_backoff_ms,
569        }
570    }
571
572    /// 业务作用:拒绝同时填写多个 provider 块或未编入的运行时能力。
573    /// 参数说明:无。
574    /// 返回:恰好一个匹配 provider 块通过。
575    pub fn validate(&self) -> Result<()> {
576        self.policy().validate()?;
577        if let Some(rest) = &self.rest {
578            rest.validate()?;
579        }
580        let blocks = [
581            self.nacos.is_some(),
582            self.consul.is_some(),
583            self.kubernetes.is_some(),
584            self.static_config.is_some(),
585            self.dns.is_some(),
586        ];
587        if blocks.into_iter().filter(|present| *present).count() != 1 {
588            return Err(invalid("one provider configuration block is required"));
589        }
590        let supported = match self.provider.as_str() {
591            "static" => self.static_config.is_some(),
592            "dns" => self.dns.is_some(),
593            "nacos" => self.nacos.is_some() && cfg!(feature = "nacos"),
594            "consul" => self.consul.is_some() && cfg!(feature = "consul"),
595            "kubernetes" => self.kubernetes.is_some() && cfg!(feature = "kubernetes"),
596            _ => false,
597        };
598        if !supported {
599            return Err(
600                invalid("provider configuration or compiled feature is unavailable")
601                    .at_config_path("provider"),
602            );
603        }
604        // 全部 provider 的配置先通过类型检查,再允许 build 读取任何认证材料。
605        if let Some(config) = &self.static_config {
606            LocalProvider::static_services(config.services.clone())?;
607        }
608        if let Some(config) = &self.dns {
609            LocalProvider::dns_services(
610                config
611                    .services
612                    .iter()
613                    .map(|(name, target)| {
614                        (
615                            name.clone(),
616                            DnsService {
617                                host: target.host.clone(),
618                                port: target.port,
619                            },
620                        )
621                    })
622                    .collect(),
623            )?;
624        }
625        #[cfg(feature = "nacos")]
626        if let Some(value) = &self.nacos {
627            let mut config: nanacos::managed::ManagedNacosConfig =
628                serde_json::from_value(value.clone())
629                    .map_err(|_| invalid("invalid Nacos provider configuration"))?;
630            config.request_timeout_ms = self.request_timeout_ms;
631            nanacos::managed::ManagedNacos::new(config)?;
632        }
633        #[cfg(feature = "consul")]
634        if let Some(value) = &self.consul {
635            let mut config: nadisc_consul::ConsulConfig = serde_json::from_value(value.clone())
636                .map_err(|_| invalid("invalid Consul provider configuration"))?;
637            config.request_timeout_ms = self.request_timeout_ms;
638            nadisc_consul::ConsulProvider::new(config)?;
639        }
640        #[cfg(feature = "kubernetes")]
641        if let Some(value) = &self.kubernetes {
642            let mut config: nadisc_kubernetes::KubernetesConfig =
643                serde_json::from_value(value.clone())
644                    .map_err(|_| invalid("invalid Kubernetes provider configuration"))?;
645            config.request_timeout_ms = self.request_timeout_ms;
646            config.validate()?;
647        }
648        Ok(())
649    }
650
651    /// 业务作用:构造只属于当前连接的后端,不复用其他账号会话。
652    /// 参数说明:无。
653    /// 返回:中性 provider;认证文件读取受其 adapter 自身有限预算约束。
654    async fn build(&self) -> Result<Arc<dyn ManagedProvider>> {
655        self.validate()?;
656        match self.provider.as_str() {
657            "static" => Ok(Arc::new(LocalProvider::static_services(
658                self.static_config.as_ref().unwrap().services.clone(),
659            )?)),
660            "dns" => Ok(Arc::new(LocalProvider::dns_services(
661                self.dns
662                    .as_ref()
663                    .unwrap()
664                    .services
665                    .iter()
666                    .map(|(name, target)| {
667                        (
668                            name.clone(),
669                            DnsService {
670                                host: target.host.clone(),
671                                port: target.port,
672                            },
673                        )
674                    })
675                    .collect(),
676            )?)),
677            #[cfg(feature = "nacos")]
678            "nacos" => {
679                let mut config: nanacos::managed::ManagedNacosConfig =
680                    serde_json::from_value(self.nacos.clone().unwrap())
681                        .map_err(|_| invalid("invalid Nacos provider configuration"))?;
682                config.request_timeout_ms = self.request_timeout_ms;
683                Ok(Arc::new(nanacos::managed::ManagedNacos::new(config)?))
684            }
685            #[cfg(feature = "consul")]
686            "consul" => {
687                let mut config: nadisc_consul::ConsulConfig =
688                    serde_json::from_value(self.consul.clone().unwrap())
689                        .map_err(|_| invalid("invalid Consul provider configuration"))?;
690                config.request_timeout_ms = self.request_timeout_ms;
691                Ok(Arc::new(nadisc_consul::ConsulProvider::new(config)?))
692            }
693            #[cfg(feature = "kubernetes")]
694            "kubernetes" => {
695                let mut config: nadisc_kubernetes::KubernetesConfig =
696                    serde_json::from_value(self.kubernetes.clone().unwrap())
697                        .map_err(|_| invalid("invalid Kubernetes provider configuration"))?;
698                config.request_timeout_ms = self.request_timeout_ms;
699                Ok(Arc::new(
700                    nadisc_kubernetes::KubernetesProvider::new(config).await?,
701                ))
702            }
703            _ => Err(invalid("provider is unavailable")),
704        }
705    }
706}
707
708/// 业务作用:保存按需编入的后端配置前递归拒绝重复键,防止秘密或作用域字段被覆盖。
709/// 参数说明:`deserializer` 必须来自尚未折叠重复项的原始输入。
710/// 返回:完整唯一键值树;重复项错误不包含原始值。
711fn optional_unique_value<'de, D: serde::Deserializer<'de>>(
712    deserializer: D,
713) -> std::result::Result<Option<serde_json::Value>, D::Error> {
714    Option::<UniqueValue>::deserialize(deserializer).map(|value| value.map(|value| value.0))
715}
716
717struct UniqueValue(serde_json::Value);
718impl<'de> Deserialize<'de> for UniqueValue {
719    /// 业务作用:在容纳后端扩展字段时保留原始重复键校验。
720    /// 参数说明:`deserializer` 提供原始序列化事件。
721    /// 返回:完整 JSON 兼容树;不接受非有限数值和重复字段。
722    fn deserialize<D: serde::Deserializer<'de>>(
723        deserializer: D,
724    ) -> std::result::Result<Self, D::Error> {
725        struct Unique;
726        impl<'de> Visitor<'de> for Unique {
727            type Value = UniqueValue;
728            /// 业务作用:返回不泄露配置内容的输入类型诊断。
729            /// 参数说明:`formatter` 为诊断目标。
730            /// 返回:固定类型描述。
731            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
732                formatter.write_str("a configuration value with unique fields")
733            }
734            /// 业务作用:校验映射每个字段只出现一次。
735            /// 参数说明:`map` 是未折叠输入。
736            /// 返回:唯一字段对象,重复输入整体失败。
737            fn visit_map<A: MapAccess<'de>>(
738                self,
739                mut map: A,
740            ) -> std::result::Result<Self::Value, A::Error> {
741                let mut values = serde_json::Map::new();
742                while let Some((key, value)) = map.next_entry::<String, UniqueValue>()? {
743                    if values.insert(key, value.0).is_some() {
744                        return Err(A::Error::custom("duplicate discovery field"));
745                    }
746                }
747                Ok(UniqueValue(serde_json::Value::Object(values)))
748            }
749            /// 业务作用:对配置数组中的每个对象继续执行字段唯一性检查。
750            /// 参数说明:`seq` 是输入数组。
751            /// 返回:完整数组,不截断元素。
752            fn visit_seq<A: serde::de::SeqAccess<'de>>(
753                self,
754                mut seq: A,
755            ) -> std::result::Result<Self::Value, A::Error> {
756                let mut values = Vec::new();
757                while let Some(value) = seq.next_element::<UniqueValue>()? {
758                    values.push(value.0);
759                }
760                Ok(UniqueValue(serde_json::Value::Array(values)))
761            }
762            /// 业务作用:保留文本配置的原始内容供后续 provider 校验。
763            /// 参数说明:`value` 是字符串值。
764            /// 返回:拥有内容的 JSON 字符串,不输出诊断日志。
765            fn visit_str<E: Error>(self, value: &str) -> std::result::Result<Self::Value, E> {
766                Ok(UniqueValue(value.into()))
767            }
768            /// 业务作用:保留布尔配置,不转换为字符串。
769            /// 参数说明:`value` 是布尔值。
770            /// 返回:对应 JSON 布尔值。
771            fn visit_bool<E: Error>(self, value: bool) -> std::result::Result<Self::Value, E> {
772                Ok(UniqueValue(value.into()))
773            }
774            /// 业务作用:保留无符号配置数值的范围。
775            /// 参数说明:`value` 是无符号整数。
776            /// 返回:对应 JSON 数字。
777            fn visit_u64<E: Error>(self, value: u64) -> std::result::Result<Self::Value, E> {
778                Ok(UniqueValue(value.into()))
779            }
780            /// 业务作用:保留有符号配置供后续范围校验。
781            /// 参数说明:`value` 是有符号整数。
782            /// 返回:对应 JSON 数字。
783            fn visit_i64<E: Error>(self, value: i64) -> std::result::Result<Self::Value, E> {
784                Ok(UniqueValue(value.into()))
785            }
786            /// 业务作用:拒绝不能表达确定权重或预算的非有限数值。
787            /// 参数说明:`value` 是浮点输入。
788            /// 返回:有限 JSON 数字或配置错误。
789            fn visit_f64<E: Error>(self, value: f64) -> std::result::Result<Self::Value, E> {
790                serde_json::Number::from_f64(value)
791                    .map(|value| UniqueValue(value.into()))
792                    .ok_or_else(|| E::custom("non-finite discovery value"))
793            }
794            /// 业务作用:保留显式空值以供类型化字段判断是否允许缺省。
795            /// 参数说明:无。
796            /// 返回:JSON 空值。
797            fn visit_unit<E: Error>(self) -> std::result::Result<Self::Value, E> {
798                Ok(UniqueValue(serde_json::Value::Null))
799            }
800        }
801        deserializer.deserialize_any(Unique)
802    }
803}
804
805/// 业务作用:在映射落地前拒绝重复名称,避免反序列化覆盖掩盖含混连接和注册。
806/// 参数说明:`deserializer` 提供原始映射事件,不接受预先丢失重复项的输入作为证明。
807/// 返回:唯一键映射;错误不输出可能包含秘密的值。
808fn unique_map<'de, D, T>(deserializer: D) -> std::result::Result<BTreeMap<String, T>, D::Error>
809where
810    D: serde::Deserializer<'de>,
811    T: Deserialize<'de>,
812{
813    struct Unique<T>(std::marker::PhantomData<T>);
814    impl<'de, T: Deserialize<'de>> Visitor<'de> for Unique<T> {
815        type Value = BTreeMap<String, T>;
816        /// 业务作用:报告固定的映射类型要求。
817        /// 参数说明:`formatter` 接收无配置值的类型描述。
818        /// 返回:安全的反序列化诊断。
819        fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
820            formatter.write_str("a map with unique names")
821        }
822        /// 业务作用:逐项检查名称唯一性后才保存配置。
823        /// 参数说明:`map` 是尚未合并的原始映射。
824        /// 返回:完整唯一键集合或重复键错误。
825        fn visit_map<A: MapAccess<'de>>(
826            self,
827            mut map: A,
828        ) -> std::result::Result<Self::Value, A::Error> {
829            let mut values = BTreeMap::new();
830            while let Some((key, value)) = map.next_entry::<String, T>()? {
831                if values.insert(key, value).is_some() {
832                    return Err(A::Error::custom("duplicate discovery name"));
833                }
834            }
835            Ok(values)
836        }
837    }
838    deserializer.deserialize_map(Unique(std::marker::PhantomData))
839}