sedsnet/
config.rs

1//! Runtime telemetry configuration and schema registry.
2//!
3//! v4 removes compile-time schema generation. `DataType` and `DataEndpoint`
4//! are stable runtime IDs, and metadata is looked up through the process-local
5//! registry. Applications may seed the registry from JSON at startup, or add
6//! endpoints/types as the network announces them.
7
8use crate::{
9    E2eEncryptionPolicy, EndpointMeta, MessageClass, MessageDataType, MessageElement, MessageMeta,
10    ReliableMode, TelemetryError, TelemetryResult, parse_f64, parse_strings, parse_usize,
11};
12#[cfg(feature = "std")]
13use alloc::sync::Arc;
14use alloc::{
15    string::{String, ToString},
16    vec::Vec,
17};
18use core::mem::size_of;
19use core::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
20
21#[cfg(feature = "std")]
22use std::sync::OnceLock;
23#[cfg(feature = "std")]
24use std::sync::RwLock;
25
26// -----------------------------------------------------------------------------
27// Device-/build-time constants
28// -----------------------------------------------------------------------------
29
30pub const DEVICE_IDENTIFIER: &str = match option_env!("DEVICE_IDENTIFIER") {
31    Some(val) => parse_strings(val),
32    None => "TEST_PLATFORM",
33};
34
35#[cfg(feature = "std")]
36static RUNTIME_DEVICE_IDENTIFIER: OnceLock<RwLock<String>> = OnceLock::new();
37
38pub fn runtime_device_identifier() -> String {
39    #[cfg(feature = "std")]
40    {
41        RUNTIME_DEVICE_IDENTIFIER
42            .get_or_init(|| RwLock::new(DEVICE_IDENTIFIER.to_string()))
43            .read()
44            .map(|value| value.clone())
45            .unwrap_or_else(|_| DEVICE_IDENTIFIER.to_string())
46    }
47    #[cfg(not(feature = "std"))]
48    {
49        DEVICE_IDENTIFIER.to_string()
50    }
51}
52
53pub fn set_runtime_device_identifier(value: &str) -> TelemetryResult<()> {
54    if value.is_empty() {
55        return Err(TelemetryError::BadArg);
56    }
57    #[cfg(feature = "std")]
58    {
59        let lock =
60            RUNTIME_DEVICE_IDENTIFIER.get_or_init(|| RwLock::new(DEVICE_IDENTIFIER.to_string()));
61        let mut guard = lock
62            .write()
63            .map_err(|_| TelemetryError::Io("device id lock"))?;
64        *guard = value.to_string();
65        Ok(())
66    }
67    #[cfg(not(feature = "std"))]
68    {
69        let _ = value;
70        Err(TelemetryError::BadArg)
71    }
72}
73
74pub const MAX_RECENT_RX_IDS: usize = match option_env!("MAX_RECENT_RX_IDS") {
75    Some(val) => parse_usize(val),
76    None => 128,
77};
78
79pub const STARTING_QUEUE_SIZE: usize = match option_env!("STARTING_QUEUE_SIZE") {
80    Some(val) => parse_usize(val),
81    None => 128,
82};
83
84pub const MAX_QUEUE_BUDGET: usize = match option_env!("MAX_QUEUE_BUDGET") {
85    Some(val) => parse_usize(val),
86    None => match option_env!("MAX_QUEUE_SIZE") {
87        Some(val) => parse_usize(val),
88        None => 1024 * 100,
89    },
90};
91
92/// Read buffer used by bounded JSON schema loading. Embedded `std` builds can
93/// lower or raise it with `SCHEMA_JSON_CHUNK_BYTES`; `no_std` builds do not use
94/// a file-read buffer.
95#[cfg(feature = "std")]
96pub const SCHEMA_JSON_CHUNK_BYTES: usize = match option_env!("SCHEMA_JSON_CHUNK_BYTES") {
97    Some(value) => {
98        let parsed = parse_usize(value);
99        if parsed == 0 { 1 } else { parsed }
100    }
101    None => 512,
102};
103
104/// Maximum JSON input size relative to the retained schema budget. This permits
105/// normal formatting overhead without allowing an attacker-controlled document
106/// to drive an unbounded parser allocation.
107#[cfg(feature = "std")]
108const SCHEMA_JSON_INPUT_MULTIPLIER: usize = 8;
109
110pub const RECENT_RX_QUEUE_BYTES: usize = {
111    let requested = MAX_RECENT_RX_IDS.saturating_mul(size_of::<u64>());
112    if requested < MAX_QUEUE_BUDGET {
113        requested
114    } else {
115        MAX_QUEUE_BUDGET
116    }
117};
118
119pub const QUEUE_GROW_STEP: f64 = match option_env!("QUEUE_GROW_STEP") {
120    Some(val) => parse_f64(val),
121    None => 3.2,
122};
123
124pub const PAYLOAD_COMPRESS_THRESHOLD: usize = match option_env!("PAYLOAD_COMPRESS_THRESHOLD") {
125    Some(val) => parse_usize(val),
126    None => 128,
127};
128
129pub const STATIC_STRING_LENGTH: usize = match option_env!("STATIC_STRING_LENGTH") {
130    Some(val) => parse_usize(val),
131    None => 1024,
132};
133
134pub const STATIC_HEX_LENGTH: usize = match option_env!("STATIC_HEX_LENGTH") {
135    Some(val) => parse_usize(val),
136    None => 1024,
137};
138
139pub const STRING_PRECISION: usize = match option_env!("STRING_PRECISION") {
140    Some(val) => parse_usize(val),
141    None => 8,
142};
143
144sedsnet_macros::define_stack_payload!(env = "MAX_STACK_PAYLOAD", default = 64);
145
146pub const MAX_HANDLER_RETRIES: usize = match option_env!("MAX_HANDLER_RETRIES") {
147    Some(val) => parse_usize(val),
148    None => 3,
149};
150
151pub const RELIABLE_RETRANSMIT_MS: u64 = match option_env!("RELIABLE_RETRANSMIT_MS") {
152    Some(val) => parse_usize(val) as u64,
153    None => 200,
154};
155
156pub const RELIABLE_MAX_RETRIES: u32 = match option_env!("RELIABLE_MAX_RETRIES") {
157    Some(val) => parse_usize(val) as u32,
158    None => 8,
159};
160
161pub const RELIABLE_MAX_PENDING: usize = match option_env!("RELIABLE_MAX_PENDING") {
162    Some(val) => parse_usize(val),
163    None => 32,
164};
165
166pub const RELIABLE_MAX_RETURN_ROUTES: usize = match option_env!("RELIABLE_MAX_RETURN_ROUTES") {
167    Some(val) => parse_usize(val),
168    None => MAX_RECENT_RX_IDS,
169};
170
171pub const RELIABLE_MAX_END_TO_END_PENDING: usize =
172    match option_env!("RELIABLE_MAX_END_TO_END_PENDING") {
173        Some(val) => parse_usize(val),
174        None => RELIABLE_MAX_PENDING,
175    };
176
177pub const RELIABLE_MAX_END_TO_END_ACK_CACHE: usize =
178    match option_env!("RELIABLE_MAX_END_TO_END_ACK_CACHE") {
179        Some(val) => parse_usize(val),
180        None => MAX_RECENT_RX_IDS,
181    };
182
183static RUNTIME_PAYLOAD_COMPRESS_THRESHOLD: AtomicUsize =
184    AtomicUsize::new(PAYLOAD_COMPRESS_THRESHOLD);
185static RUNTIME_STATIC_STRING_LENGTH: AtomicUsize = AtomicUsize::new(STATIC_STRING_LENGTH);
186static RUNTIME_STATIC_HEX_LENGTH: AtomicUsize = AtomicUsize::new(STATIC_HEX_LENGTH);
187static RUNTIME_STRING_PRECISION: AtomicUsize = AtomicUsize::new(STRING_PRECISION);
188static RUNTIME_MAX_HANDLER_RETRIES: AtomicUsize = AtomicUsize::new(MAX_HANDLER_RETRIES);
189static RUNTIME_RELIABLE_RETRANSMIT_MS: AtomicU32 = AtomicU32::new(RELIABLE_RETRANSMIT_MS as u32);
190static RUNTIME_RELIABLE_MAX_RETRIES: AtomicU32 = AtomicU32::new(RELIABLE_MAX_RETRIES);
191static RUNTIME_RELIABLE_MAX_PENDING: AtomicUsize = AtomicUsize::new(RELIABLE_MAX_PENDING);
192static RUNTIME_RELIABLE_MAX_RETURN_ROUTES: AtomicUsize =
193    AtomicUsize::new(RELIABLE_MAX_RETURN_ROUTES);
194static RUNTIME_RELIABLE_MAX_END_TO_END_PENDING: AtomicUsize =
195    AtomicUsize::new(RELIABLE_MAX_END_TO_END_PENDING);
196static RUNTIME_RELIABLE_MAX_END_TO_END_ACK_CACHE: AtomicUsize =
197    AtomicUsize::new(RELIABLE_MAX_END_TO_END_ACK_CACHE);
198
199#[derive(Debug, Clone, Copy, PartialEq, Eq)]
200pub struct RuntimeTuningConfig {
201    pub payload_compress_threshold: usize,
202    pub static_string_length: usize,
203    pub static_hex_length: usize,
204    pub string_precision: usize,
205    pub max_handler_retries: usize,
206    pub reliable_retransmit_ms: u32,
207    pub reliable_max_retries: u32,
208    pub reliable_max_pending: usize,
209    pub reliable_max_return_routes: usize,
210    pub reliable_max_end_to_end_pending: usize,
211    pub reliable_max_end_to_end_ack_cache: usize,
212}
213
214impl Default for RuntimeTuningConfig {
215    fn default() -> Self {
216        Self {
217            payload_compress_threshold: PAYLOAD_COMPRESS_THRESHOLD,
218            static_string_length: STATIC_STRING_LENGTH,
219            static_hex_length: STATIC_HEX_LENGTH,
220            string_precision: STRING_PRECISION,
221            max_handler_retries: MAX_HANDLER_RETRIES,
222            reliable_retransmit_ms: RELIABLE_RETRANSMIT_MS as u32,
223            reliable_max_retries: RELIABLE_MAX_RETRIES,
224            reliable_max_pending: RELIABLE_MAX_PENDING,
225            reliable_max_return_routes: RELIABLE_MAX_RETURN_ROUTES,
226            reliable_max_end_to_end_pending: RELIABLE_MAX_END_TO_END_PENDING,
227            reliable_max_end_to_end_ack_cache: RELIABLE_MAX_END_TO_END_ACK_CACHE,
228        }
229    }
230}
231
232impl RuntimeTuningConfig {
233    pub fn validate(self) -> TelemetryResult<()> {
234        if self.static_string_length == 0
235            || self.static_hex_length == 0
236            || self.max_handler_retries == 0
237            || self.reliable_retransmit_ms == 0
238            || self.reliable_max_retries == 0
239            || self.reliable_max_pending == 0
240            || self.reliable_max_return_routes == 0
241            || self.reliable_max_end_to_end_pending == 0
242            || self.reliable_max_end_to_end_ack_cache == 0
243        {
244            return Err(TelemetryError::BadArg);
245        }
246        Ok(())
247    }
248}
249
250pub fn set_runtime_tuning_config(cfg: RuntimeTuningConfig) -> TelemetryResult<()> {
251    cfg.validate()?;
252    RUNTIME_PAYLOAD_COMPRESS_THRESHOLD.store(cfg.payload_compress_threshold, Ordering::Relaxed);
253    RUNTIME_STATIC_STRING_LENGTH.store(cfg.static_string_length, Ordering::Relaxed);
254    RUNTIME_STATIC_HEX_LENGTH.store(cfg.static_hex_length, Ordering::Relaxed);
255    RUNTIME_STRING_PRECISION.store(cfg.string_precision, Ordering::Relaxed);
256    RUNTIME_MAX_HANDLER_RETRIES.store(cfg.max_handler_retries, Ordering::Relaxed);
257    RUNTIME_RELIABLE_RETRANSMIT_MS.store(cfg.reliable_retransmit_ms, Ordering::Relaxed);
258    RUNTIME_RELIABLE_MAX_RETRIES.store(cfg.reliable_max_retries, Ordering::Relaxed);
259    RUNTIME_RELIABLE_MAX_PENDING.store(cfg.reliable_max_pending, Ordering::Relaxed);
260    RUNTIME_RELIABLE_MAX_RETURN_ROUTES.store(cfg.reliable_max_return_routes, Ordering::Relaxed);
261    RUNTIME_RELIABLE_MAX_END_TO_END_PENDING
262        .store(cfg.reliable_max_end_to_end_pending, Ordering::Relaxed);
263    RUNTIME_RELIABLE_MAX_END_TO_END_ACK_CACHE
264        .store(cfg.reliable_max_end_to_end_ack_cache, Ordering::Relaxed);
265    Ok(())
266}
267
268pub fn runtime_tuning_config() -> RuntimeTuningConfig {
269    RuntimeTuningConfig {
270        payload_compress_threshold: runtime_payload_compress_threshold(),
271        static_string_length: runtime_static_string_length(),
272        static_hex_length: runtime_static_hex_length(),
273        string_precision: runtime_string_precision(),
274        max_handler_retries: runtime_max_handler_retries(),
275        reliable_retransmit_ms: runtime_reliable_retransmit_ms() as u32,
276        reliable_max_retries: runtime_reliable_max_retries(),
277        reliable_max_pending: runtime_reliable_max_pending(),
278        reliable_max_return_routes: runtime_reliable_max_return_routes(),
279        reliable_max_end_to_end_pending: runtime_reliable_max_end_to_end_pending(),
280        reliable_max_end_to_end_ack_cache: runtime_reliable_max_end_to_end_ack_cache(),
281    }
282}
283
284#[inline]
285pub fn runtime_payload_compress_threshold() -> usize {
286    RUNTIME_PAYLOAD_COMPRESS_THRESHOLD.load(Ordering::Relaxed)
287}
288
289#[inline]
290pub fn runtime_static_string_length() -> usize {
291    RUNTIME_STATIC_STRING_LENGTH.load(Ordering::Relaxed)
292}
293
294#[inline]
295pub fn runtime_static_hex_length() -> usize {
296    RUNTIME_STATIC_HEX_LENGTH.load(Ordering::Relaxed)
297}
298
299#[inline]
300pub fn runtime_string_precision() -> usize {
301    RUNTIME_STRING_PRECISION.load(Ordering::Relaxed)
302}
303
304#[inline]
305pub fn runtime_max_handler_retries() -> usize {
306    RUNTIME_MAX_HANDLER_RETRIES.load(Ordering::Relaxed)
307}
308
309#[inline]
310pub fn runtime_reliable_retransmit_ms() -> u64 {
311    u64::from(RUNTIME_RELIABLE_RETRANSMIT_MS.load(Ordering::Relaxed))
312}
313
314#[inline]
315pub fn runtime_reliable_max_retries() -> u32 {
316    RUNTIME_RELIABLE_MAX_RETRIES.load(Ordering::Relaxed)
317}
318
319#[inline]
320pub fn runtime_reliable_max_pending() -> usize {
321    RUNTIME_RELIABLE_MAX_PENDING.load(Ordering::Relaxed)
322}
323
324#[inline]
325pub fn runtime_reliable_max_return_routes() -> usize {
326    RUNTIME_RELIABLE_MAX_RETURN_ROUTES.load(Ordering::Relaxed)
327}
328
329#[inline]
330pub fn runtime_reliable_max_end_to_end_pending() -> usize {
331    RUNTIME_RELIABLE_MAX_END_TO_END_PENDING.load(Ordering::Relaxed)
332}
333
334#[inline]
335pub fn runtime_reliable_max_end_to_end_ack_cache() -> usize {
336    RUNTIME_RELIABLE_MAX_END_TO_END_ACK_CACHE.load(Ordering::Relaxed)
337}
338
339/// Runtime memory limits for router/relay queue-backed state.
340///
341/// Compile-time environment values remain the defaults for embedded builds, but applications using
342/// prebuilt binaries can now choose per-instance budgets at construction time.
343#[derive(Debug, Clone, Copy, PartialEq)]
344pub struct RuntimeMemoryConfig {
345    pub max_queue_budget: usize,
346    pub max_recent_rx_ids: usize,
347    pub starting_queue_size: usize,
348    pub queue_grow_step: f64,
349}
350
351impl RuntimeMemoryConfig {
352    pub const fn default_const() -> Self {
353        Self {
354            max_queue_budget: MAX_QUEUE_BUDGET,
355            max_recent_rx_ids: MAX_RECENT_RX_IDS,
356            starting_queue_size: STARTING_QUEUE_SIZE,
357            queue_grow_step: QUEUE_GROW_STEP,
358        }
359    }
360
361    pub fn new(
362        max_queue_budget: usize,
363        max_recent_rx_ids: usize,
364        starting_queue_size: usize,
365        queue_grow_step: f64,
366    ) -> TelemetryResult<Self> {
367        let cfg = Self {
368            max_queue_budget,
369            max_recent_rx_ids,
370            starting_queue_size,
371            queue_grow_step,
372        };
373        cfg.validate()?;
374        Ok(cfg)
375    }
376
377    pub fn validate(self) -> TelemetryResult<()> {
378        if self.max_queue_budget == 0 {
379            return Err(TelemetryError::BadArg);
380        }
381        if self.max_recent_rx_ids == 0 {
382            return Err(TelemetryError::BadArg);
383        }
384        if self.starting_queue_size == 0 || self.starting_queue_size > self.max_queue_budget {
385            return Err(TelemetryError::BadArg);
386        }
387        if !self.queue_grow_step.is_finite() || self.queue_grow_step <= 1.0 {
388            return Err(TelemetryError::BadArg);
389        }
390        Ok(())
391    }
392
393    pub fn recent_rx_queue_bytes(self) -> usize {
394        self.max_recent_rx_ids
395            .saturating_mul(size_of::<u64>())
396            .min(self.max_queue_budget)
397            .max(1)
398    }
399}
400
401impl Default for RuntimeMemoryConfig {
402    fn default() -> Self {
403        Self::default_const()
404    }
405}
406
407// -----------------------------------------------------------------------------
408// Runtime IDs
409// -----------------------------------------------------------------------------
410
411#[derive(Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
412#[repr(transparent)]
413pub struct DataEndpoint(pub u32);
414
415impl DataEndpoint {
416    pub const TIME_SYNC: Self = Self(200);
417    pub const DISCOVERY: Self = Self(201);
418    pub const TELEMETRY_ERROR: Self = Self(202);
419
420    #[allow(non_upper_case_globals)]
421    pub const TelemetryError: Self = Self::TELEMETRY_ERROR;
422    #[allow(non_upper_case_globals)]
423    pub const TimeSync: Self = Self::TIME_SYNC;
424    #[allow(non_upper_case_globals)]
425    pub const Discovery: Self = Self::DISCOVERY;
426
427    #[inline]
428    pub const fn as_u32(self) -> u32 {
429        self.0
430    }
431
432    #[inline]
433    pub fn try_from_u32(x: u32) -> Option<Self> {
434        if endpoint_exists(Self(x)) {
435            Some(Self(x))
436        } else {
437            None
438        }
439    }
440
441    #[inline]
442    pub fn try_named(name: &str) -> Option<Self> {
443        endpoint_definition_by_name(name).map(|def| def.id)
444    }
445
446    #[inline]
447    pub fn named(name: &str) -> Self {
448        Self::try_named(name).unwrap_or_else(|| panic!("unknown data endpoint: {name}"))
449    }
450}
451
452impl core::fmt::Debug for DataEndpoint {
453    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
454        let name = match *self {
455            Self::TelemetryError => "SEDSNET_ERROR",
456            Self::TimeSync => "SEDSNET_TIME_SYNC",
457            Self::Discovery => "SEDSNET_DISCOVERY",
458            _ => {
459                let meta = get_endpoint_meta(*self);
460                let meta_name = meta.name_ref();
461                if meta_name != "UNKNOWN_ENDPOINT" {
462                    return f.write_str(meta_name);
463                }
464                return write!(f, "DataEndpoint({})", self.0);
465            }
466        };
467        f.write_str(name)
468    }
469}
470
471#[derive(Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
472#[repr(transparent)]
473pub struct DataType(pub u32);
474
475impl DataType {
476    pub const TELEMETRY_ERROR: Self = Self(0);
477    pub const RELIABLE_ACK: Self = Self(1);
478    pub const RELIABLE_PACKET_REQUEST: Self = Self(2);
479    pub const RELIABLE_PARTIAL_ACK: Self = Self(3);
480    pub const TIME_SYNC_ANNOUNCE: Self = Self(4);
481    pub const TIME_SYNC_REQUEST: Self = Self(5);
482    pub const TIME_SYNC_RESPONSE: Self = Self(6);
483    pub const DISCOVERY_ANNOUNCE: Self = Self(7);
484    pub const DISCOVERY_TIMESYNC_SOURCES: Self = Self(8);
485    pub const DISCOVERY_TOPOLOGY: Self = Self(9);
486    pub const DISCOVERY_SCHEMA: Self = Self(10);
487    pub const DISCOVERY_TOPOLOGY_REQUEST: Self = Self(11);
488    pub const DISCOVERY_SCHEMA_REQUEST: Self = Self(12);
489    pub const MANAGED_VARIABLE_REQUEST: Self = Self(13);
490    pub const MANAGED_VARIABLE_VALUE: Self = Self(14);
491    pub const DISCOVERY_LEAVE: Self = Self(15);
492    pub const DISCOVERY_LINK_CAPABILITIES: Self = Self(16);
493    pub const DISCOVERY_ADDRESS: Self = Self(17);
494    pub const P2P_MESSAGE: Self = Self(18);
495
496    #[allow(non_upper_case_globals)]
497    pub const TelemetryError: Self = Self::TELEMETRY_ERROR;
498    #[allow(non_upper_case_globals)]
499    pub const ReliableAck: Self = Self::RELIABLE_ACK;
500    #[allow(non_upper_case_globals)]
501    pub const ReliablePacketRequest: Self = Self::RELIABLE_PACKET_REQUEST;
502    #[allow(non_upper_case_globals)]
503    pub const ReliablePartialAck: Self = Self::RELIABLE_PARTIAL_ACK;
504    #[allow(non_upper_case_globals)]
505    pub const TimeSyncAnnounce: Self = Self::TIME_SYNC_ANNOUNCE;
506    #[allow(non_upper_case_globals)]
507    pub const TimeSyncRequest: Self = Self::TIME_SYNC_REQUEST;
508    #[allow(non_upper_case_globals)]
509    pub const TimeSyncResponse: Self = Self::TIME_SYNC_RESPONSE;
510    #[allow(non_upper_case_globals)]
511    pub const DiscoveryAnnounce: Self = Self::DISCOVERY_ANNOUNCE;
512    #[allow(non_upper_case_globals)]
513    pub const DiscoveryTimeSyncSources: Self = Self::DISCOVERY_TIMESYNC_SOURCES;
514    #[allow(non_upper_case_globals)]
515    pub const DiscoveryTopology: Self = Self::DISCOVERY_TOPOLOGY;
516    #[allow(non_upper_case_globals)]
517    pub const DiscoverySchema: Self = Self::DISCOVERY_SCHEMA;
518    #[allow(non_upper_case_globals)]
519    pub const DiscoveryTopologyRequest: Self = Self::DISCOVERY_TOPOLOGY_REQUEST;
520    #[allow(non_upper_case_globals)]
521    pub const DiscoverySchemaRequest: Self = Self::DISCOVERY_SCHEMA_REQUEST;
522    #[allow(non_upper_case_globals)]
523    pub const ManagedVariableRequest: Self = Self::MANAGED_VARIABLE_REQUEST;
524    #[allow(non_upper_case_globals)]
525    pub const ManagedVariableValue: Self = Self::MANAGED_VARIABLE_VALUE;
526    #[allow(non_upper_case_globals)]
527    pub const DiscoveryLeave: Self = Self::DISCOVERY_LEAVE;
528    #[allow(non_upper_case_globals)]
529    pub const DiscoveryLinkCapabilities: Self = Self::DISCOVERY_LINK_CAPABILITIES;
530    #[allow(non_upper_case_globals)]
531    pub const DiscoveryAddress: Self = Self::DISCOVERY_ADDRESS;
532    #[allow(non_upper_case_globals)]
533    pub const P2pMessage: Self = Self::P2P_MESSAGE;
534
535    #[inline]
536    pub const fn as_u32(self) -> u32 {
537        self.0
538    }
539
540    #[inline]
541    pub fn try_from_u32(x: u32) -> Option<Self> {
542        if data_type_exists(Self(x)) {
543            Some(Self(x))
544        } else {
545            None
546        }
547    }
548
549    #[inline]
550    pub fn try_named(name: &str) -> Option<Self> {
551        data_type_definition_by_name(name).map(|def| def.id)
552    }
553
554    #[inline]
555    pub fn named(name: &str) -> Self {
556        Self::try_named(name).unwrap_or_else(|| panic!("unknown data type: {name}"))
557    }
558}
559
560impl core::fmt::Debug for DataType {
561    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
562        let name = match *self {
563            Self::TelemetryError => "SEDSNET_ERROR",
564            Self::ReliableAck => "ReliableAck",
565            Self::ReliablePacketRequest => "ReliablePacketRequest",
566            Self::ReliablePartialAck => "ReliablePartialAck",
567            Self::TimeSyncAnnounce => "SedsnetTimeSyncAnnounce",
568            Self::TimeSyncRequest => "SedsnetTimeSyncRequest",
569            Self::TimeSyncResponse => "SedsnetTimeSyncResponse",
570            Self::DiscoveryAnnounce => "SedsnetDiscoveryAnnounce",
571            Self::DiscoveryTimeSyncSources => "SedsnetDiscoveryTimeSyncSources",
572            Self::DiscoveryTopology => "SedsnetDiscoveryTopology",
573            Self::DiscoverySchema => "SedsnetDiscoverySchema",
574            Self::DiscoveryTopologyRequest => "SedsnetDiscoveryTopologyRequest",
575            Self::DiscoverySchemaRequest => "SedsnetDiscoverySchemaRequest",
576            Self::ManagedVariableRequest => "SedsnetManagedVariableRequest",
577            Self::ManagedVariableValue => "SedsnetManagedVariableValue",
578            Self::DiscoveryLeave => "SedsnetDiscoveryLeave",
579            Self::DiscoveryLinkCapabilities => "SedsnetDiscoveryLinkCapabilities",
580            Self::DiscoveryAddress => "SedsnetDiscoveryAddress",
581            Self::P2pMessage => "SedsnetP2pMessage",
582            _ => {
583                let meta = get_message_meta(*self);
584                let meta_name = meta.name_ref();
585                if meta_name != "UNKNOWN_TYPE" {
586                    return f.write_str(meta_name);
587                }
588                return write!(f, "DataType({})", self.0);
589            }
590        };
591        f.write_str(name)
592    }
593}
594
595// -----------------------------------------------------------------------------
596// Runtime registry
597// -----------------------------------------------------------------------------
598
599#[derive(Debug, Clone, Copy, PartialEq, Eq)]
600pub struct EndpointDefinition {
601    pub id: DataEndpoint,
602    pub name: &'static str,
603    pub description: &'static str,
604    pub link_local_only: bool,
605}
606
607#[derive(Debug, Clone, Copy, PartialEq, Eq)]
608pub struct DataTypeDefinition {
609    pub id: DataType,
610    pub name: &'static str,
611    pub description: &'static str,
612    pub element: MessageElement,
613    pub endpoints: &'static [DataEndpoint],
614    pub reliable: ReliableMode,
615    pub priority: u8,
616    pub e2e_encryption: E2eEncryptionPolicy,
617}
618
619#[cfg(not(feature = "std"))]
620include!(concat!(env!("OUT_DIR"), "/embedded_schema.rs"));
621
622#[derive(Debug, Clone)]
623pub struct RuntimeSchemaSnapshot {
624    pub endpoints: Vec<EndpointDefinition>,
625    pub types: Vec<DataTypeDefinition>,
626}
627
628#[derive(Debug, Clone, PartialEq, Eq)]
629pub struct OwnedEndpointDefinition {
630    pub id: DataEndpoint,
631    pub name: String,
632    pub description: String,
633    pub link_local_only: bool,
634}
635
636#[derive(Debug, Clone, PartialEq, Eq)]
637pub struct OwnedDataTypeDefinition {
638    pub id: DataType,
639    pub name: String,
640    pub description: String,
641    pub element: MessageElement,
642    pub endpoints: Vec<DataEndpoint>,
643    pub reliable: ReliableMode,
644    pub priority: u8,
645    pub e2e_encryption: E2eEncryptionPolicy,
646}
647
648#[derive(Debug, Clone)]
649pub struct OwnedRuntimeSchemaSnapshot {
650    pub endpoints: Vec<OwnedEndpointDefinition>,
651    pub types: Vec<OwnedDataTypeDefinition>,
652}
653
654impl PartialEq<EndpointDefinition> for OwnedEndpointDefinition {
655    fn eq(&self, other: &EndpointDefinition) -> bool {
656        self.id == other.id
657            && self.name == other.name
658            && self.description == other.description
659            && self.link_local_only == other.link_local_only
660    }
661}
662
663#[derive(Debug, Clone, Copy, PartialEq, Eq)]
664pub enum SchemaMergeDecision {
665    Added,
666    Unchanged,
667    ReplacedLocal,
668    KeptLocal,
669}
670
671#[derive(Debug, Clone, Copy, PartialEq, Eq)]
672pub struct SchemaMergeReport {
673    pub endpoints_added: usize,
674    pub endpoints_replaced: usize,
675    pub endpoints_kept: usize,
676    pub types_added: usize,
677    pub types_replaced: usize,
678    pub types_kept: usize,
679}
680
681impl SchemaMergeReport {
682    #[inline]
683    pub const fn changed(&self) -> bool {
684        self.endpoints_added != 0
685            || self.endpoints_replaced != 0
686            || self.types_added != 0
687            || self.types_replaced != 0
688    }
689}
690
691#[cfg(feature = "std")]
692#[derive(Debug, Clone)]
693struct Registry {
694    endpoints: Vec<(DataEndpoint, EndpointMeta)>,
695    types: Vec<(DataType, MessageMeta)>,
696    next_endpoint_id: u32,
697    next_type_id: u32,
698}
699
700#[cfg(feature = "std")]
701impl Registry {
702    fn new() -> Self {
703        let mut reg = Self {
704            endpoints: Vec::new(),
705            types: Vec::new(),
706            next_endpoint_id: 100,
707            next_type_id: 100,
708        };
709        reg.register_endpoint_definition(EndpointDefinition {
710            id: DataEndpoint::TelemetryError,
711            name: "SEDSNET_ERROR",
712            description: "",
713            link_local_only: false,
714        })
715        .expect("built-in endpoint");
716        reg.register_endpoint_definition(EndpointDefinition {
717            id: DataEndpoint::TimeSync,
718            name: "SEDSNET_TIME_SYNC",
719            description: "",
720            link_local_only: false,
721        })
722        .expect("built-in endpoint");
723        reg.register_endpoint_definition(EndpointDefinition {
724            id: DataEndpoint::Discovery,
725            name: "SEDSNET_DISCOVERY",
726            description: "",
727            link_local_only: false,
728        })
729        .expect("built-in endpoint");
730
731        reg.register_type_definition(DataTypeDefinition {
732            id: DataType::TelemetryError,
733            name: "SEDSNET_ERROR",
734            description: "",
735            element: MessageElement::Dynamic(MessageDataType::String, MessageClass::Error),
736            endpoints: &[DataEndpoint::TelemetryError],
737            reliable: ReliableMode::None,
738            priority: 255,
739            e2e_encryption: E2eEncryptionPolicy::PreferOff,
740        })
741        .expect("built-in type");
742        reg.register_type_definition(DataTypeDefinition {
743            id: DataType::ReliableAck,
744            name: "SEDSNET_RELIABLE_ACK",
745            description: "",
746            element: MessageElement::Static(2, MessageDataType::UInt32, MessageClass::Data),
747            endpoints: &[DataEndpoint::TelemetryError],
748            reliable: ReliableMode::None,
749            priority: 250,
750            e2e_encryption: E2eEncryptionPolicy::PreferOff,
751        })
752        .expect("built-in type");
753        reg.register_type_definition(DataTypeDefinition {
754            id: DataType::ReliablePacketRequest,
755            name: "SEDSNET_RELIABLE_PACKET_REQUEST",
756            description: "",
757            element: MessageElement::Static(2, MessageDataType::UInt32, MessageClass::Data),
758            endpoints: &[DataEndpoint::TelemetryError],
759            reliable: ReliableMode::None,
760            priority: 250,
761            e2e_encryption: E2eEncryptionPolicy::PreferOff,
762        })
763        .expect("built-in type");
764        reg.register_type_definition(DataTypeDefinition {
765            id: DataType::ReliablePartialAck,
766            name: "SEDSNET_RELIABLE_PARTIAL_ACK",
767            description: "",
768            element: MessageElement::Static(2, MessageDataType::UInt32, MessageClass::Data),
769            endpoints: &[DataEndpoint::TelemetryError],
770            reliable: ReliableMode::None,
771            priority: 250,
772            e2e_encryption: E2eEncryptionPolicy::PreferOff,
773        })
774        .expect("built-in type");
775        reg.register_type_definition(DataTypeDefinition {
776            id: DataType::TimeSyncAnnounce,
777            name: "SEDSNET_TIME_SYNC_ANNOUNCE",
778            description: "",
779            element: MessageElement::Static(2, MessageDataType::UInt64, MessageClass::Data),
780            endpoints: &[DataEndpoint::TimeSync],
781            reliable: ReliableMode::None,
782            priority: 245,
783            e2e_encryption: E2eEncryptionPolicy::PreferOff,
784        })
785        .expect("built-in type");
786        reg.register_type_definition(DataTypeDefinition {
787            id: DataType::TimeSyncRequest,
788            name: "SEDSNET_TIME_SYNC_REQUEST",
789            description: "",
790            element: MessageElement::Static(2, MessageDataType::UInt64, MessageClass::Data),
791            endpoints: &[DataEndpoint::TimeSync],
792            reliable: ReliableMode::None,
793            priority: 245,
794            e2e_encryption: E2eEncryptionPolicy::PreferOff,
795        })
796        .expect("built-in type");
797        reg.register_type_definition(DataTypeDefinition {
798            id: DataType::TimeSyncResponse,
799            name: "SEDSNET_TIME_SYNC_RESPONSE",
800            description: "",
801            element: MessageElement::Static(4, MessageDataType::UInt64, MessageClass::Data),
802            endpoints: &[DataEndpoint::TimeSync],
803            reliable: ReliableMode::None,
804            priority: 245,
805            e2e_encryption: E2eEncryptionPolicy::PreferOff,
806        })
807        .expect("built-in type");
808        reg.register_type_definition(DataTypeDefinition {
809            id: DataType::DiscoveryAnnounce,
810            name: "SEDSNET_DISCOVERY_ANNOUNCE",
811            description: "",
812            element: MessageElement::Dynamic(MessageDataType::UInt32, MessageClass::Data),
813            endpoints: &[DataEndpoint::Discovery],
814            reliable: ReliableMode::None,
815            priority: 240,
816            e2e_encryption: E2eEncryptionPolicy::PreferOff,
817        })
818        .expect("built-in type");
819        reg.register_type_definition(DataTypeDefinition {
820            id: DataType::DiscoveryTimeSyncSources,
821            name: "SEDSNET_DISCOVERY_TIMESYNC_SOURCES",
822            description: "",
823            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
824            endpoints: &[DataEndpoint::Discovery],
825            reliable: ReliableMode::None,
826            priority: 240,
827            e2e_encryption: E2eEncryptionPolicy::PreferOff,
828        })
829        .expect("built-in type");
830        reg.register_type_definition(DataTypeDefinition {
831            id: DataType::DiscoveryTopology,
832            name: "SEDSNET_DISCOVERY_TOPOLOGY",
833            description: "",
834            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
835            endpoints: &[DataEndpoint::Discovery],
836            reliable: ReliableMode::Ordered,
837            priority: 240,
838            e2e_encryption: E2eEncryptionPolicy::PreferOff,
839        })
840        .expect("built-in type");
841        reg.register_type_definition(DataTypeDefinition {
842            id: DataType::DiscoverySchema,
843            name: "SEDSNET_DISCOVERY_SCHEMA",
844            description: "",
845            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
846            endpoints: &[DataEndpoint::Discovery],
847            reliable: ReliableMode::Ordered,
848            priority: 241,
849            e2e_encryption: E2eEncryptionPolicy::PreferOff,
850        })
851        .expect("built-in type");
852        reg.register_type_definition(DataTypeDefinition {
853            id: DataType::DiscoveryTopologyRequest,
854            name: "SEDSNET_DISCOVERY_TOPOLOGY_REQUEST",
855            description: "",
856            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
857            endpoints: &[DataEndpoint::Discovery],
858            reliable: ReliableMode::Ordered,
859            priority: 242,
860            e2e_encryption: E2eEncryptionPolicy::PreferOff,
861        })
862        .expect("built-in type");
863        reg.register_type_definition(DataTypeDefinition {
864            id: DataType::DiscoverySchemaRequest,
865            name: "SEDSNET_DISCOVERY_SCHEMA_REQUEST",
866            description: "",
867            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
868            endpoints: &[DataEndpoint::Discovery],
869            reliable: ReliableMode::Ordered,
870            priority: 242,
871            e2e_encryption: E2eEncryptionPolicy::PreferOff,
872        })
873        .expect("built-in type");
874        reg.register_type_definition(DataTypeDefinition {
875            id: DataType::ManagedVariableRequest,
876            name: "SEDSNET_MANAGED_VARIABLE_REQUEST",
877            description: "",
878            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
879            endpoints: &[DataEndpoint::Discovery],
880            reliable: ReliableMode::Ordered,
881            priority: 243,
882            e2e_encryption: E2eEncryptionPolicy::PreferOff,
883        })
884        .expect("built-in type");
885        reg.register_type_definition(DataTypeDefinition {
886            id: DataType::ManagedVariableValue,
887            name: "SEDSNET_MANAGED_VARIABLE_VALUE",
888            description: "",
889            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
890            endpoints: &[DataEndpoint::Discovery],
891            reliable: ReliableMode::Ordered,
892            priority: 243,
893            e2e_encryption: E2eEncryptionPolicy::PreferOff,
894        })
895        .expect("built-in type");
896        reg.register_type_definition(DataTypeDefinition {
897            id: DataType::DiscoveryLeave,
898            name: "SEDSNET_DISCOVERY_LEAVE",
899            description: "",
900            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
901            endpoints: &[DataEndpoint::Discovery],
902            reliable: ReliableMode::None,
903            priority: 244,
904            e2e_encryption: E2eEncryptionPolicy::PreferOff,
905        })
906        .expect("built-in type");
907        reg.register_type_definition(DataTypeDefinition {
908            id: DataType::DiscoveryLinkCapabilities,
909            name: "SEDSNET_DISCOVERY_LINK_CAPABILITIES",
910            description: "",
911            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
912            endpoints: &[DataEndpoint::Discovery],
913            reliable: ReliableMode::None,
914            priority: 240,
915            e2e_encryption: E2eEncryptionPolicy::PreferOff,
916        })
917        .expect("built-in type");
918        reg.register_type_definition(DataTypeDefinition {
919            id: DataType::DiscoveryAddress,
920            name: "SEDSNET_DISCOVERY_ADDRESS",
921            description: "",
922            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
923            endpoints: &[DataEndpoint::Discovery],
924            reliable: ReliableMode::Ordered,
925            priority: 244,
926            e2e_encryption: E2eEncryptionPolicy::PreferOff,
927        })
928        .expect("built-in type");
929        reg.register_type_definition(DataTypeDefinition {
930            id: DataType::P2pMessage,
931            name: "SEDSNET_P2P_MESSAGE",
932            description: "",
933            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
934            endpoints: &[DataEndpoint::Discovery],
935            reliable: ReliableMode::Ordered,
936            priority: 246,
937            e2e_encryption: E2eEncryptionPolicy::PreferOff,
938        })
939        .expect("built-in type");
940        #[cfg(all(feature = "embedded", sedsnet_has_telemetry_config_json))]
941        if let Ok(snapshot) = bundled_schema_snapshot() {
942            let _ = register_owned_schema_snapshot_into(&mut reg, snapshot);
943        }
944        let _ = register_runtime_json_config(
945            &mut reg,
946            "SEDSNET_STATIC_SCHEMA_PATH",
947            false,
948            MAX_QUEUE_BUDGET,
949        );
950        let _ = register_runtime_json_config(
951            &mut reg,
952            "SEDSNET_STATIC_IPC_SCHEMA_PATH",
953            true,
954            MAX_QUEUE_BUDGET,
955        );
956        reg
957    }
958
959    fn register_endpoint_definition(&mut self, def: EndpointDefinition) -> TelemetryResult<()> {
960        self.register_owned_endpoint(OwnedEndpointDefinition {
961            id: def.id,
962            name: def.name.to_string(),
963            description: def.description.to_string(),
964            link_local_only: def.link_local_only,
965        })
966    }
967
968    fn register_owned_endpoint(&mut self, def: OwnedEndpointDefinition) -> TelemetryResult<()> {
969        if let Some((_, existing)) = self.endpoints.iter().find(|(id, _)| *id == def.id) {
970            if existing.name.as_ref() == def.name
971                && existing.description.as_ref() == def.description
972                && existing.link_local_only == def.link_local_only
973            {
974                return Ok(());
975            }
976            return Err(TelemetryError::BadArg);
977        }
978        if self
979            .endpoints
980            .iter()
981            .any(|(_, meta)| meta.name.as_ref() == def.name)
982        {
983            return Err(TelemetryError::BadArg);
984        }
985        self.next_endpoint_id = self.next_endpoint_id.max(def.id.0.saturating_add(1));
986        self.endpoints.push((
987            def.id,
988            EndpointMeta {
989                name: Arc::from(def.name),
990                description: Arc::from(def.description),
991                link_local_only: def.link_local_only,
992            },
993        ));
994        self.endpoints.sort_unstable_by_key(|(id, _)| id.0);
995        Ok(())
996    }
997
998    fn register_type_definition(&mut self, def: DataTypeDefinition) -> TelemetryResult<()> {
999        self.register_owned_type(OwnedDataTypeDefinition {
1000            id: def.id,
1001            name: def.name.to_string(),
1002            description: def.description.to_string(),
1003            element: def.element,
1004            endpoints: def.endpoints.to_vec(),
1005            reliable: def.reliable,
1006            priority: def.priority,
1007            e2e_encryption: def.e2e_encryption,
1008        })
1009    }
1010
1011    fn register_owned_type(&mut self, def: OwnedDataTypeDefinition) -> TelemetryResult<()> {
1012        if let Some((_, existing)) = self.types.iter().find(|(id, _)| *id == def.id) {
1013            if existing.name.as_ref() == def.name
1014                && existing.description.as_ref() == def.description
1015                && existing.element == def.element
1016                && existing.endpoints.as_ref() == def.endpoints
1017                && existing.reliable == def.reliable
1018                && existing.priority == def.priority
1019                && existing.e2e_encryption == def.e2e_encryption
1020            {
1021                return Ok(());
1022            }
1023            return Err(TelemetryError::BadArg);
1024        }
1025        if self
1026            .types
1027            .iter()
1028            .any(|(_, meta)| meta.name.as_ref() == def.name)
1029        {
1030            return Err(TelemetryError::BadArg);
1031        }
1032        for ep in &def.endpoints {
1033            if !self.endpoints.iter().any(|(id, _)| id == ep) {
1034                return Err(TelemetryError::BadArg);
1035            }
1036        }
1037        self.next_type_id = self.next_type_id.max(def.id.0.saturating_add(1));
1038        self.types.push((
1039            def.id,
1040            MessageMeta {
1041                name: Arc::from(def.name),
1042                description: Arc::from(def.description),
1043                element: def.element,
1044                endpoints: Arc::from(def.endpoints),
1045                reliable: def.reliable,
1046                priority: def.priority,
1047                e2e_encryption: def.e2e_encryption,
1048            },
1049        ));
1050        self.types.sort_unstable_by_key(|(id, _)| id.0);
1051        Ok(())
1052    }
1053
1054    fn schema_byte_cost(&self) -> usize {
1055        self.endpoints
1056            .iter()
1057            .map(|(_, meta)| endpoint_schema_byte_cost(meta.name.len(), meta.description.len()))
1058            .sum::<usize>()
1059            .saturating_add(
1060                self.types
1061                    .iter()
1062                    .map(|(_, meta)| {
1063                        type_schema_byte_cost(
1064                            meta.name.len(),
1065                            meta.description.len(),
1066                            meta.endpoints.len(),
1067                        )
1068                    })
1069                    .sum::<usize>(),
1070            )
1071    }
1072
1073    fn merge_endpoint_definition(&mut self, def: OwnedEndpointDefinition) -> SchemaMergeDecision {
1074        let id_match = self.endpoints.iter().position(|(id, _)| *id == def.id);
1075        let name_match = self
1076            .endpoints
1077            .iter()
1078            .position(|(_, meta)| meta.name.as_ref() == def.name);
1079        let conflict = match (id_match, name_match) {
1080            (Some(a), Some(b)) if a != b => Some(a.min(b)),
1081            (Some(a), _) | (_, Some(a)) => Some(a),
1082            (None, None) => None,
1083        };
1084
1085        let Some(idx) = conflict else {
1086            self.next_endpoint_id = self.next_endpoint_id.max(def.id.0.saturating_add(1));
1087            self.endpoints.push((
1088                def.id,
1089                EndpointMeta {
1090                    name: Arc::from(def.name),
1091                    description: Arc::from(def.description),
1092                    link_local_only: def.link_local_only,
1093                },
1094            ));
1095            self.endpoints.sort_unstable_by_key(|(id, _)| id.0);
1096            return SchemaMergeDecision::Added;
1097        };
1098
1099        let existing = self.endpoints[idx].clone();
1100        let existing_def = OwnedEndpointDefinition {
1101            id: existing.0,
1102            name: existing.1.name.to_string(),
1103            description: existing.1.description.to_string(),
1104            link_local_only: existing.1.link_local_only,
1105        };
1106        if endpoint_def_equivalent(&existing_def, &def) {
1107            return SchemaMergeDecision::Unchanged;
1108        }
1109        if endpoint_winner(&existing_def, &def) == def {
1110            self.endpoints[idx] = (
1111                def.id,
1112                EndpointMeta {
1113                    name: Arc::from(def.name),
1114                    description: Arc::from(def.description),
1115                    link_local_only: def.link_local_only,
1116                },
1117            );
1118            self.endpoints.sort_unstable_by_key(|(id, _)| id.0);
1119            self.next_endpoint_id = self.next_endpoint_id.max(def.id.0.saturating_add(1));
1120            SchemaMergeDecision::ReplacedLocal
1121        } else {
1122            SchemaMergeDecision::KeptLocal
1123        }
1124    }
1125
1126    fn merge_type_definition(&mut self, def: OwnedDataTypeDefinition) -> SchemaMergeDecision {
1127        let id_match = self.types.iter().position(|(id, _)| *id == def.id);
1128        let name_match = self
1129            .types
1130            .iter()
1131            .position(|(_, meta)| meta.name.as_ref() == def.name);
1132        let conflict = match (id_match, name_match) {
1133            (Some(a), Some(b)) if a != b => Some(a.min(b)),
1134            (Some(a), _) | (_, Some(a)) => Some(a),
1135            (None, None) => None,
1136        };
1137
1138        let Some(idx) = conflict else {
1139            self.next_type_id = self.next_type_id.max(def.id.0.saturating_add(1));
1140            self.types.push((
1141                def.id,
1142                MessageMeta {
1143                    name: Arc::from(def.name),
1144                    description: Arc::from(def.description),
1145                    element: def.element,
1146                    endpoints: Arc::from(def.endpoints),
1147                    reliable: def.reliable,
1148                    priority: def.priority,
1149                    e2e_encryption: def.e2e_encryption,
1150                },
1151            ));
1152            self.types.sort_unstable_by_key(|(id, _)| id.0);
1153            return SchemaMergeDecision::Added;
1154        };
1155
1156        let existing = self.types[idx].clone();
1157        let existing_def = OwnedDataTypeDefinition {
1158            id: existing.0,
1159            name: existing.1.name.to_string(),
1160            description: existing.1.description.to_string(),
1161            element: existing.1.element,
1162            endpoints: existing.1.endpoints.to_vec(),
1163            reliable: existing.1.reliable,
1164            priority: existing.1.priority,
1165            e2e_encryption: existing.1.e2e_encryption,
1166        };
1167        if type_def_equivalent(&existing_def, &def) {
1168            return SchemaMergeDecision::Unchanged;
1169        }
1170        if type_winner(&existing_def, &def) == def {
1171            self.types[idx] = (
1172                def.id,
1173                MessageMeta {
1174                    name: Arc::from(def.name),
1175                    description: Arc::from(def.description),
1176                    element: def.element,
1177                    endpoints: Arc::from(def.endpoints),
1178                    reliable: def.reliable,
1179                    priority: def.priority,
1180                    e2e_encryption: def.e2e_encryption,
1181                },
1182            );
1183            self.types.sort_unstable_by_key(|(id, _)| id.0);
1184            self.next_type_id = self.next_type_id.max(def.id.0.saturating_add(1));
1185            SchemaMergeDecision::ReplacedLocal
1186        } else {
1187            SchemaMergeDecision::KeptLocal
1188        }
1189    }
1190}
1191
1192fn endpoint_schema_byte_cost(name_len: usize, description_len: usize) -> usize {
1193    size_of::<(DataEndpoint, EndpointMeta)>()
1194        .saturating_add(name_len)
1195        .saturating_add(description_len)
1196}
1197
1198fn type_schema_byte_cost(name_len: usize, description_len: usize, endpoint_count: usize) -> usize {
1199    size_of::<(DataType, MessageMeta)>()
1200        .saturating_add(name_len)
1201        .saturating_add(description_len)
1202        .saturating_add(endpoint_count.saturating_mul(size_of::<DataEndpoint>()))
1203}
1204
1205pub fn owned_schema_byte_cost(snapshot: &OwnedRuntimeSchemaSnapshot) -> usize {
1206    snapshot
1207        .endpoints
1208        .iter()
1209        .map(|def| endpoint_schema_byte_cost(def.name.len(), def.description.len()))
1210        .sum::<usize>()
1211        .saturating_add(
1212            snapshot
1213                .types
1214                .iter()
1215                .map(|def| {
1216                    type_schema_byte_cost(
1217                        def.name.len(),
1218                        def.description.len(),
1219                        def.endpoints.len(),
1220                    )
1221                })
1222                .sum::<usize>(),
1223        )
1224}
1225
1226#[cfg(feature = "std")]
1227static REGISTRY: OnceLock<std::sync::Mutex<Registry>> = OnceLock::new();
1228
1229#[cfg(feature = "std")]
1230fn registry() -> &'static std::sync::Mutex<Registry> {
1231    REGISTRY.get_or_init(|| std::sync::Mutex::new(Registry::new()))
1232}
1233
1234#[cfg(all(
1235    feature = "std",
1236    feature = "serde",
1237    feature = "embedded",
1238    sedsnet_has_telemetry_config_json
1239))]
1240fn bundled_schema_snapshot() -> TelemetryResult<OwnedRuntimeSchemaSnapshot> {
1241    schema_snapshot_from_json_bytes(include_bytes!("../telemetry_config.json"))
1242}
1243
1244#[cfg(feature = "std")]
1245fn register_runtime_json_config(
1246    reg: &mut Registry,
1247    env_key: &str,
1248    link_local_overlay: bool,
1249    max_schema_bytes: usize,
1250) -> TelemetryResult<()> {
1251    let path = std::env::var(env_key).map_err(|_| TelemetryError::Io("schema json path"))?;
1252    register_schema_json_file_into(
1253        reg,
1254        std::path::Path::new(&path),
1255        link_local_overlay,
1256        max_schema_bytes,
1257        SCHEMA_JSON_CHUNK_BYTES,
1258    )
1259}
1260
1261#[cfg(feature = "std")]
1262pub fn register_endpoint(name: &str, link_local_only: bool) -> TelemetryResult<DataEndpoint> {
1263    register_endpoint_with_description(name, "", link_local_only)
1264}
1265
1266#[cfg(feature = "std")]
1267pub fn register_endpoint_with_description(
1268    name: &str,
1269    description: &str,
1270    link_local_only: bool,
1271) -> TelemetryResult<DataEndpoint> {
1272    let mut reg = registry().lock().expect("schema registry poisoned");
1273    let id = DataEndpoint(reg.next_endpoint_id);
1274    reg.register_owned_endpoint(OwnedEndpointDefinition {
1275        id,
1276        name: name.to_string(),
1277        description: description.to_string(),
1278        link_local_only,
1279    })?;
1280    Ok(id)
1281}
1282
1283#[cfg(feature = "std")]
1284pub fn register_endpoint_id(
1285    id: DataEndpoint,
1286    name: &str,
1287    link_local_only: bool,
1288) -> TelemetryResult<DataEndpoint> {
1289    register_endpoint_id_with_description(id, name, "", link_local_only)
1290}
1291
1292#[cfg(feature = "std")]
1293pub fn register_endpoint_id_with_description(
1294    id: DataEndpoint,
1295    name: &str,
1296    description: &str,
1297    link_local_only: bool,
1298) -> TelemetryResult<DataEndpoint> {
1299    registry()
1300        .lock()
1301        .expect("schema registry poisoned")
1302        .register_owned_endpoint(OwnedEndpointDefinition {
1303            id,
1304            name: name.to_string(),
1305            description: description.to_string(),
1306            link_local_only,
1307        })?;
1308    Ok(id)
1309}
1310
1311#[cfg(feature = "std")]
1312pub fn ensure_endpoint_id(
1313    id: DataEndpoint,
1314    link_local_only: bool,
1315) -> TelemetryResult<DataEndpoint> {
1316    if endpoint_exists(id) {
1317        return Ok(id);
1318    }
1319    register_endpoint_id(id, &format!("ENDPOINT_{}", id.0), link_local_only)
1320}
1321
1322#[cfg(feature = "std")]
1323pub fn register_endpoint_definition(def: EndpointDefinition) -> TelemetryResult<()> {
1324    registry()
1325        .lock()
1326        .expect("schema registry poisoned")
1327        .register_endpoint_definition(def)
1328}
1329
1330#[cfg(feature = "std")]
1331pub fn register_data_type(
1332    name: &str,
1333    element: MessageElement,
1334    endpoints: &[DataEndpoint],
1335    reliable: ReliableMode,
1336    priority: u8,
1337) -> TelemetryResult<DataType> {
1338    register_data_type_with_description(name, "", element, endpoints, reliable, priority)
1339}
1340
1341#[cfg(feature = "std")]
1342pub fn register_data_type_with_description(
1343    name: &str,
1344    description: &str,
1345    element: MessageElement,
1346    endpoints: &[DataEndpoint],
1347    reliable: ReliableMode,
1348    priority: u8,
1349) -> TelemetryResult<DataType> {
1350    register_data_type_with_description_and_e2e_encryption(
1351        name,
1352        description,
1353        element,
1354        endpoints,
1355        reliable,
1356        priority,
1357        E2eEncryptionPolicy::PreferOff,
1358    )
1359}
1360
1361#[cfg(feature = "std")]
1362#[allow(clippy::too_many_arguments)]
1363pub fn register_data_type_with_description_and_e2e_encryption(
1364    name: &str,
1365    description: &str,
1366    element: MessageElement,
1367    endpoints: &[DataEndpoint],
1368    reliable: ReliableMode,
1369    priority: u8,
1370    e2e_encryption: E2eEncryptionPolicy,
1371) -> TelemetryResult<DataType> {
1372    let mut reg = registry().lock().expect("schema registry poisoned");
1373    let id = DataType(reg.next_type_id);
1374    reg.register_owned_type(OwnedDataTypeDefinition {
1375        id,
1376        name: name.to_string(),
1377        description: description.to_string(),
1378        element,
1379        endpoints: endpoints.to_vec(),
1380        reliable,
1381        priority,
1382        e2e_encryption,
1383    })?;
1384    Ok(id)
1385}
1386
1387#[cfg(feature = "std")]
1388pub fn register_data_type_definition(def: DataTypeDefinition) -> TelemetryResult<()> {
1389    registry()
1390        .lock()
1391        .expect("schema registry poisoned")
1392        .register_type_definition(def)
1393}
1394
1395#[cfg(feature = "std")]
1396pub fn set_data_type_e2e_encryption_policy(
1397    ty: DataType,
1398    policy: E2eEncryptionPolicy,
1399) -> TelemetryResult<()> {
1400    let mut reg = registry().lock().expect("schema registry poisoned");
1401    let Some((_, meta)) = reg.types.iter_mut().find(|(id, _)| *id == ty) else {
1402        return Err(TelemetryError::InvalidType);
1403    };
1404    meta.e2e_encryption = policy;
1405    Ok(())
1406}
1407
1408#[cfg(feature = "std")]
1409pub fn register_data_type_id(
1410    id: DataType,
1411    name: &str,
1412    element: MessageElement,
1413    endpoints: &[DataEndpoint],
1414    reliable: ReliableMode,
1415    priority: u8,
1416) -> TelemetryResult<DataType> {
1417    register_data_type_id_with_description(id, name, "", element, endpoints, reliable, priority)
1418}
1419
1420#[cfg(feature = "std")]
1421pub fn register_data_type_id_with_description(
1422    id: DataType,
1423    name: &str,
1424    description: &str,
1425    element: MessageElement,
1426    endpoints: &[DataEndpoint],
1427    reliable: ReliableMode,
1428    priority: u8,
1429) -> TelemetryResult<DataType> {
1430    register_data_type_id_with_description_and_e2e_encryption(
1431        id,
1432        name,
1433        description,
1434        element,
1435        endpoints,
1436        reliable,
1437        priority,
1438        E2eEncryptionPolicy::PreferOff,
1439    )
1440}
1441
1442#[cfg(feature = "std")]
1443#[allow(clippy::too_many_arguments)]
1444pub fn register_data_type_id_with_description_and_e2e_encryption(
1445    id: DataType,
1446    name: &str,
1447    description: &str,
1448    element: MessageElement,
1449    endpoints: &[DataEndpoint],
1450    reliable: ReliableMode,
1451    priority: u8,
1452    e2e_encryption: E2eEncryptionPolicy,
1453) -> TelemetryResult<DataType> {
1454    registry()
1455        .lock()
1456        .expect("schema registry poisoned")
1457        .register_owned_type(OwnedDataTypeDefinition {
1458            id,
1459            name: name.to_string(),
1460            description: description.to_string(),
1461            element,
1462            endpoints: endpoints.to_vec(),
1463            reliable,
1464            priority,
1465            e2e_encryption,
1466        })?;
1467    Ok(id)
1468}
1469
1470#[cfg(feature = "std")]
1471pub fn merge_schema_snapshot(snapshot: RuntimeSchemaSnapshot) -> SchemaMergeReport {
1472    merge_owned_schema_snapshot(OwnedRuntimeSchemaSnapshot {
1473        endpoints: snapshot
1474            .endpoints
1475            .into_iter()
1476            .map(|def| OwnedEndpointDefinition {
1477                id: def.id,
1478                name: def.name.to_string(),
1479                description: def.description.to_string(),
1480                link_local_only: def.link_local_only,
1481            })
1482            .collect(),
1483        types: snapshot
1484            .types
1485            .into_iter()
1486            .map(|def| OwnedDataTypeDefinition {
1487                id: def.id,
1488                name: def.name.to_string(),
1489                description: def.description.to_string(),
1490                element: def.element,
1491                endpoints: def.endpoints.to_vec(),
1492                reliable: def.reliable,
1493                priority: def.priority,
1494                e2e_encryption: def.e2e_encryption,
1495            })
1496            .collect(),
1497    })
1498}
1499
1500#[cfg(feature = "std")]
1501pub fn merge_owned_schema_snapshot(snapshot: OwnedRuntimeSchemaSnapshot) -> SchemaMergeReport {
1502    merge_owned_schema_snapshot_with_budget(snapshot, usize::MAX)
1503        .expect("unbounded schema merge should not fail budget")
1504}
1505
1506#[cfg(feature = "std")]
1507pub fn merge_owned_schema_snapshot_with_budget(
1508    mut snapshot: OwnedRuntimeSchemaSnapshot,
1509    max_schema_bytes: usize,
1510) -> TelemetryResult<SchemaMergeReport> {
1511    snapshot.endpoints.sort_unstable_by_key(|def| def.id.0);
1512    snapshot.endpoints.dedup_by_key(|def| def.id.0);
1513    snapshot.types.sort_unstable_by_key(|def| def.id.0);
1514    snapshot.types.dedup_by_key(|def| def.id.0);
1515
1516    let reg = registry().lock().expect("schema registry poisoned");
1517    if reg
1518        .schema_byte_cost()
1519        .saturating_add(owned_schema_byte_cost(&snapshot))
1520        > max_schema_bytes
1521    {
1522        return Err(TelemetryError::PacketTooLarge(
1523            "Schema exceeds maximum shared queue budget",
1524        ));
1525    }
1526    drop(reg);
1527
1528    let mut reg = registry().lock().expect("schema registry poisoned");
1529    let mut preview = reg.clone();
1530    let report = merge_owned_schema_snapshot_locked(&mut preview, snapshot);
1531    if preview.schema_byte_cost() > max_schema_bytes {
1532        return Err(TelemetryError::PacketTooLarge(
1533            "Schema exceeds maximum shared queue budget",
1534        ));
1535    }
1536    *reg = preview;
1537    Ok(report)
1538}
1539
1540#[cfg(feature = "std")]
1541fn merge_owned_schema_snapshot_locked(
1542    reg: &mut Registry,
1543    mut snapshot: OwnedRuntimeSchemaSnapshot,
1544) -> SchemaMergeReport {
1545    snapshot.endpoints.sort_unstable_by_key(|def| def.id.0);
1546    snapshot.endpoints.dedup_by_key(|def| def.id.0);
1547    snapshot.types.sort_unstable_by_key(|def| def.id.0);
1548    snapshot.types.dedup_by_key(|def| def.id.0);
1549
1550    let mut report = SchemaMergeReport {
1551        endpoints_added: 0,
1552        endpoints_replaced: 0,
1553        endpoints_kept: 0,
1554        types_added: 0,
1555        types_replaced: 0,
1556        types_kept: 0,
1557    };
1558    for endpoint in snapshot.endpoints {
1559        match reg.merge_endpoint_definition(endpoint) {
1560            SchemaMergeDecision::Added => report.endpoints_added += 1,
1561            SchemaMergeDecision::ReplacedLocal => report.endpoints_replaced += 1,
1562            SchemaMergeDecision::KeptLocal => report.endpoints_kept += 1,
1563            SchemaMergeDecision::Unchanged => {}
1564        }
1565    }
1566    for ty in snapshot.types {
1567        if ty
1568            .endpoints
1569            .iter()
1570            .all(|ep| reg.endpoints.iter().any(|(known_ep, _)| known_ep == ep))
1571        {
1572            match reg.merge_type_definition(ty) {
1573                SchemaMergeDecision::Added => report.types_added += 1,
1574                SchemaMergeDecision::ReplacedLocal => report.types_replaced += 1,
1575                SchemaMergeDecision::KeptLocal => report.types_kept += 1,
1576                SchemaMergeDecision::Unchanged => {}
1577            }
1578        } else {
1579            report.types_kept += 1;
1580        }
1581    }
1582    report
1583}
1584
1585#[cfg(feature = "std")]
1586pub fn export_schema() -> OwnedRuntimeSchemaSnapshot {
1587    let reg = registry().lock().expect("schema registry poisoned");
1588    OwnedRuntimeSchemaSnapshot {
1589        endpoints: reg
1590            .endpoints
1591            .iter()
1592            .map(|(id, meta)| OwnedEndpointDefinition {
1593                id: *id,
1594                name: meta.name.to_string(),
1595                description: meta.description.to_string(),
1596                link_local_only: meta.link_local_only,
1597            })
1598            .collect(),
1599        types: reg
1600            .types
1601            .iter()
1602            .map(|(id, meta)| OwnedDataTypeDefinition {
1603                id: *id,
1604                name: meta.name.to_string(),
1605                description: meta.description.to_string(),
1606                element: meta.element,
1607                endpoints: meta.endpoints.to_vec(),
1608                reliable: meta.reliable,
1609                priority: meta.priority,
1610                e2e_encryption: meta.e2e_encryption,
1611            })
1612            .collect(),
1613    }
1614}
1615
1616#[cfg(feature = "std")]
1617pub fn known_endpoints() -> Vec<OwnedEndpointDefinition> {
1618    export_schema().endpoints
1619}
1620
1621#[cfg(feature = "std")]
1622pub fn known_data_types() -> Vec<OwnedDataTypeDefinition> {
1623    export_schema().types
1624}
1625
1626#[cfg(feature = "std")]
1627pub fn schema_fingerprint() -> u64 {
1628    let snapshot = export_schema();
1629    let mut h = 0x5E_D5_50_4F_52_49_4E_54u64;
1630    for ep in snapshot.endpoints {
1631        h = hash_u32(h, ep.id.0);
1632        h = hash_bytes(h, ep.name.as_bytes());
1633        h = hash_bytes(h, ep.description.as_bytes());
1634        h = hash_u8(h, ep.link_local_only as u8);
1635    }
1636    for ty in snapshot.types {
1637        h = hash_u32(h, ty.id.0);
1638        h = hash_bytes(h, ty.name.as_bytes());
1639        h = hash_bytes(h, ty.description.as_bytes());
1640        h = hash_message_element(h, ty.element);
1641        h = hash_u8(h, reliable_code(ty.reliable));
1642        h = hash_u8(h, ty.priority);
1643        for ep in ty.endpoints {
1644            h = hash_u32(h, ep.0);
1645        }
1646    }
1647    h
1648}
1649
1650#[cfg(feature = "std")]
1651pub fn schema_bytes_used() -> usize {
1652    registry()
1653        .lock()
1654        .expect("schema registry poisoned")
1655        .schema_byte_cost()
1656}
1657
1658#[cfg(feature = "std")]
1659pub fn endpoint_exists(ep: DataEndpoint) -> bool {
1660    #[cfg(all(test, feature = "std"))]
1661    seed_test_schema();
1662    registry()
1663        .lock()
1664        .expect("schema registry poisoned")
1665        .endpoints
1666        .iter()
1667        .any(|(id, _)| *id == ep)
1668}
1669
1670#[cfg(feature = "std")]
1671pub fn data_type_exists(ty: DataType) -> bool {
1672    #[cfg(all(test, feature = "std"))]
1673    seed_test_schema();
1674    registry()
1675        .lock()
1676        .expect("schema registry poisoned")
1677        .types
1678        .iter()
1679        .any(|(id, _)| *id == ty)
1680}
1681
1682#[cfg(feature = "std")]
1683pub fn get_endpoint_meta(endpoint_type: DataEndpoint) -> EndpointMeta {
1684    #[cfg(all(test, feature = "std"))]
1685    seed_test_schema();
1686    registry()
1687        .lock()
1688        .expect("schema registry poisoned")
1689        .endpoints
1690        .iter()
1691        .find(|(id, _)| *id == endpoint_type)
1692        .map(|(_, meta)| meta.clone())
1693        .unwrap_or(EndpointMeta {
1694            name: Arc::from("UNKNOWN_ENDPOINT"),
1695            description: Arc::from(""),
1696            link_local_only: false,
1697        })
1698}
1699
1700#[cfg(feature = "std")]
1701pub fn get_message_meta(data_type: DataType) -> MessageMeta {
1702    #[cfg(all(test, feature = "std"))]
1703    seed_test_schema();
1704    registry()
1705        .lock()
1706        .expect("schema registry poisoned")
1707        .types
1708        .iter()
1709        .find(|(id, _)| *id == data_type)
1710        .map(|(_, meta)| meta.clone())
1711        .unwrap_or(MessageMeta {
1712            name: Arc::from("UNKNOWN_TYPE"),
1713            description: Arc::from(""),
1714            element: MessageElement::Dynamic(MessageDataType::Binary, MessageClass::Data),
1715            endpoints: Arc::from([]),
1716            reliable: ReliableMode::None,
1717            priority: 0,
1718            e2e_encryption: E2eEncryptionPolicy::PreferOff,
1719        })
1720}
1721
1722#[cfg(feature = "std")]
1723pub fn max_endpoint_id() -> u32 {
1724    registry()
1725        .lock()
1726        .expect("schema registry poisoned")
1727        .endpoints
1728        .iter()
1729        .map(|(id, _)| id.0)
1730        .max()
1731        .unwrap_or(0)
1732}
1733
1734#[cfg(feature = "std")]
1735pub fn max_data_type_id() -> u32 {
1736    registry()
1737        .lock()
1738        .expect("schema registry poisoned")
1739        .types
1740        .iter()
1741        .map(|(id, _)| id.0)
1742        .max()
1743        .unwrap_or(0)
1744}
1745
1746#[cfg(feature = "std")]
1747fn hash_u8(h: u64, v: u8) -> u64 {
1748    hash_bytes(h, &[v])
1749}
1750
1751#[cfg(feature = "std")]
1752fn hash_u32(h: u64, v: u32) -> u64 {
1753    hash_bytes(h, &v.to_le_bytes())
1754}
1755
1756#[cfg(feature = "std")]
1757fn hash_usize(h: u64, v: usize) -> u64 {
1758    hash_bytes(h, &(v as u64).to_le_bytes())
1759}
1760
1761#[cfg(feature = "std")]
1762fn hash_bytes(mut h: u64, bytes: &[u8]) -> u64 {
1763    const PRIME: u64 = 0x0000_0100_0000_01B3;
1764    for &b in bytes {
1765        h ^= b as u64;
1766        h = h.wrapping_mul(PRIME);
1767    }
1768    h
1769}
1770
1771#[cfg(feature = "std")]
1772fn endpoint_fingerprint(def: &OwnedEndpointDefinition) -> u64 {
1773    let mut h = 0x4550_4445_4600_0001;
1774    h = hash_u32(h, def.id.0);
1775    h = hash_bytes(h, def.name.as_bytes());
1776    h = hash_bytes(h, def.description.as_bytes());
1777    hash_u8(h, def.link_local_only as u8)
1778}
1779
1780#[cfg(feature = "std")]
1781fn type_fingerprint(def: &OwnedDataTypeDefinition) -> u64 {
1782    let mut h = 0x5459_4445_4600_0001;
1783    h = hash_u32(h, def.id.0);
1784    h = hash_bytes(h, def.name.as_bytes());
1785    h = hash_bytes(h, def.description.as_bytes());
1786    h = hash_message_element(h, def.element);
1787    h = hash_u8(h, reliable_code(def.reliable));
1788    h = hash_u8(h, def.priority);
1789    h = hash_u8(h, e2e_encryption_policy_code(def.e2e_encryption));
1790    for ep in &def.endpoints {
1791        h = hash_u32(h, ep.0);
1792    }
1793    h
1794}
1795
1796#[cfg(feature = "std")]
1797fn hash_message_element(mut h: u64, element: MessageElement) -> u64 {
1798    match element {
1799        MessageElement::Static(count, data_type, class) => {
1800            h = hash_u8(h, 0);
1801            h = hash_usize(h, count);
1802            h = hash_u8(h, message_data_type_code(data_type));
1803            hash_u8(h, message_class_code(class))
1804        }
1805        MessageElement::Dynamic(data_type, class) => {
1806            h = hash_u8(h, 1);
1807            h = hash_u8(h, message_data_type_code(data_type));
1808            hash_u8(h, message_class_code(class))
1809        }
1810    }
1811}
1812
1813#[cfg(feature = "std")]
1814pub fn endpoint_definition(ep: DataEndpoint) -> Option<OwnedEndpointDefinition> {
1815    registry()
1816        .lock()
1817        .expect("schema registry poisoned")
1818        .endpoints
1819        .iter()
1820        .find(|(id, _)| *id == ep)
1821        .map(|(id, meta)| OwnedEndpointDefinition {
1822            id: *id,
1823            name: meta.name.to_string(),
1824            description: meta.description.to_string(),
1825            link_local_only: meta.link_local_only,
1826        })
1827}
1828
1829#[cfg(feature = "std")]
1830pub fn data_type_definition(ty: DataType) -> Option<OwnedDataTypeDefinition> {
1831    registry()
1832        .lock()
1833        .expect("schema registry poisoned")
1834        .types
1835        .iter()
1836        .find(|(id, _)| *id == ty)
1837        .map(|(id, meta)| OwnedDataTypeDefinition {
1838            id: *id,
1839            name: meta.name.to_string(),
1840            description: meta.description.to_string(),
1841            element: meta.element,
1842            endpoints: meta.endpoints.to_vec(),
1843            reliable: meta.reliable,
1844            priority: meta.priority,
1845            e2e_encryption: meta.e2e_encryption,
1846        })
1847}
1848
1849#[cfg(feature = "std")]
1850pub fn endpoint_definition_by_name(name: &str) -> Option<OwnedEndpointDefinition> {
1851    registry()
1852        .lock()
1853        .expect("schema registry poisoned")
1854        .endpoints
1855        .iter()
1856        .find(|(_, meta)| meta.name.as_ref() == name)
1857        .map(|(id, meta)| OwnedEndpointDefinition {
1858            id: *id,
1859            name: meta.name.to_string(),
1860            description: meta.description.to_string(),
1861            link_local_only: meta.link_local_only,
1862        })
1863}
1864
1865#[cfg(feature = "std")]
1866pub fn data_type_definition_by_name(name: &str) -> Option<OwnedDataTypeDefinition> {
1867    registry()
1868        .lock()
1869        .expect("schema registry poisoned")
1870        .types
1871        .iter()
1872        .find(|(_, meta)| meta.name.as_ref() == name)
1873        .map(|(id, meta)| OwnedDataTypeDefinition {
1874            id: *id,
1875            name: meta.name.to_string(),
1876            description: meta.description.to_string(),
1877            element: meta.element,
1878            endpoints: meta.endpoints.to_vec(),
1879            reliable: meta.reliable,
1880            priority: meta.priority,
1881            e2e_encryption: meta.e2e_encryption,
1882        })
1883}
1884
1885#[cfg(feature = "std")]
1886fn is_internal_endpoint(ep: DataEndpoint) -> bool {
1887    matches!(
1888        ep,
1889        DataEndpoint::TelemetryError | DataEndpoint::TimeSync | DataEndpoint::Discovery
1890    )
1891}
1892
1893#[cfg(feature = "std")]
1894fn is_internal_data_type(ty: DataType) -> bool {
1895    matches!(
1896        ty,
1897        DataType::TelemetryError
1898            | DataType::ReliableAck
1899            | DataType::ReliablePacketRequest
1900            | DataType::ReliablePartialAck
1901            | DataType::TimeSyncAnnounce
1902            | DataType::TimeSyncRequest
1903            | DataType::TimeSyncResponse
1904            | DataType::DiscoveryAnnounce
1905            | DataType::DiscoveryTimeSyncSources
1906            | DataType::DiscoveryTopology
1907            | DataType::DiscoverySchema
1908            | DataType::DiscoveryTopologyRequest
1909            | DataType::DiscoverySchemaRequest
1910            | DataType::ManagedVariableRequest
1911            | DataType::ManagedVariableValue
1912            | DataType::DiscoveryLeave
1913            | DataType::DiscoveryLinkCapabilities
1914            | DataType::DiscoveryAddress
1915            | DataType::P2pMessage
1916    )
1917}
1918
1919#[cfg(feature = "std")]
1920pub fn remove_endpoint(ep: DataEndpoint) -> TelemetryResult<bool> {
1921    if is_internal_endpoint(ep) {
1922        return Err(TelemetryError::BadArg);
1923    }
1924    let mut reg = registry().lock().expect("schema registry poisoned");
1925    let before = reg.endpoints.len();
1926    reg.endpoints.retain(|(id, _)| *id != ep);
1927    if reg.endpoints.len() == before {
1928        return Ok(false);
1929    }
1930    reg.types.retain(|(_, meta)| !meta.endpoints.contains(&ep));
1931    Ok(true)
1932}
1933
1934#[cfg(feature = "std")]
1935pub fn remove_endpoint_by_name(name: &str) -> TelemetryResult<bool> {
1936    if let Some(def) = endpoint_definition_by_name(name) {
1937        remove_endpoint(def.id)
1938    } else {
1939        Ok(false)
1940    }
1941}
1942
1943#[cfg(feature = "std")]
1944pub fn remove_data_type(ty: DataType) -> TelemetryResult<bool> {
1945    if is_internal_data_type(ty) {
1946        return Err(TelemetryError::BadArg);
1947    }
1948    let mut reg = registry().lock().expect("schema registry poisoned");
1949    let before = reg.types.len();
1950    reg.types.retain(|(id, _)| *id != ty);
1951    Ok(reg.types.len() != before)
1952}
1953
1954#[cfg(feature = "std")]
1955pub fn remove_data_type_by_name(name: &str) -> TelemetryResult<bool> {
1956    if let Some(def) = data_type_definition_by_name(name) {
1957        remove_data_type(def.id)
1958    } else {
1959        Ok(false)
1960    }
1961}
1962
1963#[cfg(feature = "std")]
1964fn endpoint_def_equivalent(a: &OwnedEndpointDefinition, b: &OwnedEndpointDefinition) -> bool {
1965    a.id == b.id
1966        && a.name == b.name
1967        && a.description == b.description
1968        && a.link_local_only == b.link_local_only
1969}
1970
1971#[cfg(feature = "std")]
1972fn type_def_equivalent(a: &OwnedDataTypeDefinition, b: &OwnedDataTypeDefinition) -> bool {
1973    a.id == b.id
1974        && a.name == b.name
1975        && a.description == b.description
1976        && a.element == b.element
1977        && a.endpoints == b.endpoints
1978        && a.reliable == b.reliable
1979        && a.priority == b.priority
1980}
1981
1982#[cfg(feature = "std")]
1983fn endpoint_winner(
1984    a: &OwnedEndpointDefinition,
1985    b: &OwnedEndpointDefinition,
1986) -> OwnedEndpointDefinition {
1987    let a_key = (endpoint_fingerprint(a), a.id.0, a.name.as_str());
1988    let b_key = (endpoint_fingerprint(b), b.id.0, b.name.as_str());
1989    if a_key <= b_key { a.clone() } else { b.clone() }
1990}
1991
1992#[cfg(feature = "std")]
1993fn type_winner(
1994    a: &OwnedDataTypeDefinition,
1995    b: &OwnedDataTypeDefinition,
1996) -> OwnedDataTypeDefinition {
1997    let a_key = (type_fingerprint(a), a.id.0, a.name.as_str());
1998    let b_key = (type_fingerprint(b), b.id.0, b.name.as_str());
1999    if a_key <= b_key { a.clone() } else { b.clone() }
2000}
2001
2002pub(crate) fn message_data_type_code(dt: MessageDataType) -> u8 {
2003    match dt {
2004        MessageDataType::Float64 => 0,
2005        MessageDataType::Float32 => 1,
2006        MessageDataType::UInt8 => 2,
2007        MessageDataType::UInt16 => 3,
2008        MessageDataType::UInt32 => 4,
2009        MessageDataType::UInt64 => 5,
2010        MessageDataType::UInt128 => 6,
2011        MessageDataType::Int8 => 7,
2012        MessageDataType::Int16 => 8,
2013        MessageDataType::Int32 => 9,
2014        MessageDataType::Int64 => 10,
2015        MessageDataType::Int128 => 11,
2016        MessageDataType::Bool => 12,
2017        MessageDataType::String => 13,
2018        MessageDataType::Binary => 14,
2019        MessageDataType::NoData => 15,
2020    }
2021}
2022
2023pub(crate) fn message_data_type_from_code(code: u8) -> Option<MessageDataType> {
2024    match code {
2025        0 => Some(MessageDataType::Float64),
2026        1 => Some(MessageDataType::Float32),
2027        2 => Some(MessageDataType::UInt8),
2028        3 => Some(MessageDataType::UInt16),
2029        4 => Some(MessageDataType::UInt32),
2030        5 => Some(MessageDataType::UInt64),
2031        6 => Some(MessageDataType::UInt128),
2032        7 => Some(MessageDataType::Int8),
2033        8 => Some(MessageDataType::Int16),
2034        9 => Some(MessageDataType::Int32),
2035        10 => Some(MessageDataType::Int64),
2036        11 => Some(MessageDataType::Int128),
2037        12 => Some(MessageDataType::Bool),
2038        13 => Some(MessageDataType::String),
2039        14 => Some(MessageDataType::Binary),
2040        15 => Some(MessageDataType::NoData),
2041        _ => None,
2042    }
2043}
2044
2045pub(crate) fn message_class_code(class: MessageClass) -> u8 {
2046    match class {
2047        MessageClass::Data => 0,
2048        MessageClass::Error => 1,
2049        MessageClass::Warning => 2,
2050    }
2051}
2052
2053pub(crate) fn message_class_from_code(code: u8) -> Option<MessageClass> {
2054    match code {
2055        0 => Some(MessageClass::Data),
2056        1 => Some(MessageClass::Error),
2057        2 => Some(MessageClass::Warning),
2058        _ => None,
2059    }
2060}
2061
2062pub(crate) fn reliable_code(mode: ReliableMode) -> u8 {
2063    match mode {
2064        ReliableMode::None => 0,
2065        ReliableMode::Ordered => 1,
2066        ReliableMode::Unordered => 2,
2067    }
2068}
2069
2070pub(crate) fn reliable_from_code(code: u8) -> Option<ReliableMode> {
2071    match code {
2072        0 => Some(ReliableMode::None),
2073        1 => Some(ReliableMode::Ordered),
2074        2 => Some(ReliableMode::Unordered),
2075        _ => None,
2076    }
2077}
2078
2079pub(crate) fn e2e_encryption_policy_code(policy: E2eEncryptionPolicy) -> u8 {
2080    match policy {
2081        E2eEncryptionPolicy::PreferOff => 0,
2082        E2eEncryptionPolicy::PreferOn => 1,
2083        E2eEncryptionPolicy::RequireOn => 2,
2084    }
2085}
2086
2087pub(crate) fn e2e_encryption_policy_from_code(code: u8) -> Option<E2eEncryptionPolicy> {
2088    match code {
2089        0 => Some(E2eEncryptionPolicy::PreferOff),
2090        1 => Some(E2eEncryptionPolicy::PreferOn),
2091        2 => Some(E2eEncryptionPolicy::RequireOn),
2092        _ => None,
2093    }
2094}
2095
2096#[cfg(not(feature = "std"))]
2097pub fn register_endpoint(_name: &str, _link_local_only: bool) -> TelemetryResult<DataEndpoint> {
2098    Err(TelemetryError::BadArg)
2099}
2100
2101#[cfg(not(feature = "std"))]
2102pub fn register_endpoint_with_description(
2103    _name: &str,
2104    _description: &str,
2105    _link_local_only: bool,
2106) -> TelemetryResult<DataEndpoint> {
2107    Err(TelemetryError::BadArg)
2108}
2109
2110#[cfg(not(feature = "std"))]
2111pub fn register_endpoint_definition(_def: EndpointDefinition) -> TelemetryResult<()> {
2112    Err(TelemetryError::BadArg)
2113}
2114
2115#[cfg(not(feature = "std"))]
2116pub fn register_endpoint_id(
2117    _id: DataEndpoint,
2118    _name: &str,
2119    _link_local_only: bool,
2120) -> TelemetryResult<DataEndpoint> {
2121    Err(TelemetryError::BadArg)
2122}
2123
2124#[cfg(not(feature = "std"))]
2125pub fn register_endpoint_id_with_description(
2126    _id: DataEndpoint,
2127    _name: &str,
2128    _description: &str,
2129    _link_local_only: bool,
2130) -> TelemetryResult<DataEndpoint> {
2131    Err(TelemetryError::BadArg)
2132}
2133
2134#[cfg(not(feature = "std"))]
2135pub fn ensure_endpoint_id(
2136    id: DataEndpoint,
2137    _link_local_only: bool,
2138) -> TelemetryResult<DataEndpoint> {
2139    if endpoint_exists(id) {
2140        Ok(id)
2141    } else {
2142        Err(TelemetryError::BadArg)
2143    }
2144}
2145
2146#[cfg(not(feature = "std"))]
2147pub fn register_data_type(
2148    _name: &str,
2149    _element: MessageElement,
2150    _endpoints: &[DataEndpoint],
2151    _reliable: ReliableMode,
2152    _priority: u8,
2153) -> TelemetryResult<DataType> {
2154    Err(TelemetryError::BadArg)
2155}
2156
2157#[cfg(not(feature = "std"))]
2158pub fn register_data_type_with_description(
2159    _name: &str,
2160    _description: &str,
2161    _element: MessageElement,
2162    _endpoints: &[DataEndpoint],
2163    _reliable: ReliableMode,
2164    _priority: u8,
2165) -> TelemetryResult<DataType> {
2166    Err(TelemetryError::BadArg)
2167}
2168
2169#[cfg(not(feature = "std"))]
2170pub fn register_data_type_definition(_def: DataTypeDefinition) -> TelemetryResult<()> {
2171    Err(TelemetryError::BadArg)
2172}
2173
2174#[cfg(not(feature = "std"))]
2175pub fn set_data_type_e2e_encryption_policy(
2176    _ty: DataType,
2177    _policy: E2eEncryptionPolicy,
2178) -> TelemetryResult<()> {
2179    Err(TelemetryError::BadArg)
2180}
2181
2182#[cfg(not(feature = "std"))]
2183pub fn register_data_type_id(
2184    _id: DataType,
2185    _name: &str,
2186    _element: MessageElement,
2187    _endpoints: &[DataEndpoint],
2188    _reliable: ReliableMode,
2189    _priority: u8,
2190) -> TelemetryResult<DataType> {
2191    Err(TelemetryError::BadArg)
2192}
2193
2194#[cfg(not(feature = "std"))]
2195pub fn register_data_type_id_with_description(
2196    _id: DataType,
2197    _name: &str,
2198    _description: &str,
2199    _element: MessageElement,
2200    _endpoints: &[DataEndpoint],
2201    _reliable: ReliableMode,
2202    _priority: u8,
2203) -> TelemetryResult<DataType> {
2204    Err(TelemetryError::BadArg)
2205}
2206
2207#[cfg(not(feature = "std"))]
2208#[allow(clippy::too_many_arguments)]
2209pub fn register_data_type_id_with_description_and_e2e_encryption(
2210    _id: DataType,
2211    _name: &str,
2212    _description: &str,
2213    _element: MessageElement,
2214    _endpoints: &[DataEndpoint],
2215    _reliable: ReliableMode,
2216    _priority: u8,
2217    _e2e_encryption: E2eEncryptionPolicy,
2218) -> TelemetryResult<DataType> {
2219    Err(TelemetryError::BadArg)
2220}
2221
2222#[cfg(not(feature = "std"))]
2223pub fn export_schema() -> RuntimeSchemaSnapshot {
2224    RuntimeSchemaSnapshot {
2225        endpoints: known_endpoints(),
2226        types: known_data_types(),
2227    }
2228}
2229
2230#[cfg(not(feature = "std"))]
2231pub fn known_endpoints() -> Vec<EndpointDefinition> {
2232    let mut endpoints = Vec::with_capacity(3 + EMBEDDED_SCHEMA_ENDPOINTS.len());
2233    endpoints.extend_from_slice(&[
2234        EndpointDefinition {
2235            id: DataEndpoint::TelemetryError,
2236            name: "SEDSNET_ERROR",
2237            description: "",
2238            link_local_only: false,
2239        },
2240        EndpointDefinition {
2241            id: DataEndpoint::TimeSync,
2242            name: "SEDSNET_TIME_SYNC",
2243            description: "",
2244            link_local_only: false,
2245        },
2246        EndpointDefinition {
2247            id: DataEndpoint::Discovery,
2248            name: "SEDSNET_DISCOVERY",
2249            description: "",
2250            link_local_only: false,
2251        },
2252    ]);
2253    endpoints.extend_from_slice(EMBEDDED_SCHEMA_ENDPOINTS);
2254    endpoints
2255}
2256
2257#[cfg(not(feature = "std"))]
2258pub fn known_data_types() -> Vec<DataTypeDefinition> {
2259    let mut types = Vec::with_capacity(20 + EMBEDDED_SCHEMA_TYPES.len());
2260    types.extend_from_slice(&[
2261        DataTypeDefinition {
2262            id: DataType::TelemetryError,
2263            name: "SEDSNET_ERROR",
2264            description: "",
2265            element: MessageElement::Dynamic(MessageDataType::String, MessageClass::Error),
2266            endpoints: &[DataEndpoint::TelemetryError],
2267            reliable: ReliableMode::None,
2268            priority: 255,
2269            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2270        },
2271        DataTypeDefinition {
2272            id: DataType::ReliableAck,
2273            name: "SEDSNET_RELIABLE_ACK",
2274            description: "",
2275            element: MessageElement::Static(2, MessageDataType::UInt32, MessageClass::Data),
2276            endpoints: &[DataEndpoint::TelemetryError],
2277            reliable: ReliableMode::None,
2278            priority: 250,
2279            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2280        },
2281        DataTypeDefinition {
2282            id: DataType::ReliablePacketRequest,
2283            name: "SEDSNET_RELIABLE_PACKET_REQUEST",
2284            description: "",
2285            element: MessageElement::Static(2, MessageDataType::UInt32, MessageClass::Data),
2286            endpoints: &[DataEndpoint::TelemetryError],
2287            reliable: ReliableMode::None,
2288            priority: 250,
2289            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2290        },
2291        DataTypeDefinition {
2292            id: DataType::ReliablePartialAck,
2293            name: "SEDSNET_RELIABLE_PARTIAL_ACK",
2294            description: "",
2295            element: MessageElement::Static(2, MessageDataType::UInt32, MessageClass::Data),
2296            endpoints: &[DataEndpoint::TelemetryError],
2297            reliable: ReliableMode::None,
2298            priority: 250,
2299            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2300        },
2301        DataTypeDefinition {
2302            id: DataType::TimeSyncAnnounce,
2303            name: "SEDSNET_TIME_SYNC_ANNOUNCE",
2304            description: "",
2305            element: MessageElement::Static(2, MessageDataType::UInt64, MessageClass::Data),
2306            endpoints: &[DataEndpoint::TimeSync],
2307            reliable: ReliableMode::None,
2308            priority: 245,
2309            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2310        },
2311        DataTypeDefinition {
2312            id: DataType::TimeSyncRequest,
2313            name: "SEDSNET_TIME_SYNC_REQUEST",
2314            description: "",
2315            element: MessageElement::Static(2, MessageDataType::UInt64, MessageClass::Data),
2316            endpoints: &[DataEndpoint::TimeSync],
2317            reliable: ReliableMode::None,
2318            priority: 245,
2319            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2320        },
2321        DataTypeDefinition {
2322            id: DataType::TimeSyncResponse,
2323            name: "SEDSNET_TIME_SYNC_RESPONSE",
2324            description: "",
2325            element: MessageElement::Static(4, MessageDataType::UInt64, MessageClass::Data),
2326            endpoints: &[DataEndpoint::TimeSync],
2327            reliable: ReliableMode::None,
2328            priority: 245,
2329            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2330        },
2331        DataTypeDefinition {
2332            id: DataType::DiscoveryAnnounce,
2333            name: "SEDSNET_DISCOVERY_ANNOUNCE",
2334            description: "",
2335            element: MessageElement::Dynamic(MessageDataType::UInt32, MessageClass::Data),
2336            endpoints: &[DataEndpoint::Discovery],
2337            reliable: ReliableMode::None,
2338            priority: 240,
2339            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2340        },
2341        DataTypeDefinition {
2342            id: DataType::DiscoveryTimeSyncSources,
2343            name: "SEDSNET_DISCOVERY_TIMESYNC_SOURCES",
2344            description: "",
2345            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2346            endpoints: &[DataEndpoint::Discovery],
2347            reliable: ReliableMode::None,
2348            priority: 240,
2349            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2350        },
2351        DataTypeDefinition {
2352            id: DataType::DiscoveryTopology,
2353            name: "SEDSNET_DISCOVERY_TOPOLOGY",
2354            description: "",
2355            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2356            endpoints: &[DataEndpoint::Discovery],
2357            reliable: ReliableMode::Ordered,
2358            priority: 240,
2359            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2360        },
2361        DataTypeDefinition {
2362            id: DataType::DiscoverySchema,
2363            name: "SEDSNET_DISCOVERY_SCHEMA",
2364            description: "",
2365            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2366            endpoints: &[DataEndpoint::Discovery],
2367            reliable: ReliableMode::Ordered,
2368            priority: 241,
2369            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2370        },
2371        DataTypeDefinition {
2372            id: DataType::DiscoveryTopologyRequest,
2373            name: "SEDSNET_DISCOVERY_TOPOLOGY_REQUEST",
2374            description: "",
2375            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2376            endpoints: &[DataEndpoint::Discovery],
2377            reliable: ReliableMode::Ordered,
2378            priority: 242,
2379            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2380        },
2381        DataTypeDefinition {
2382            id: DataType::DiscoverySchemaRequest,
2383            name: "SEDSNET_DISCOVERY_SCHEMA_REQUEST",
2384            description: "",
2385            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2386            endpoints: &[DataEndpoint::Discovery],
2387            reliable: ReliableMode::Ordered,
2388            priority: 242,
2389            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2390        },
2391        DataTypeDefinition {
2392            id: DataType::ManagedVariableRequest,
2393            name: "SEDSNET_MANAGED_VARIABLE_REQUEST",
2394            description: "",
2395            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2396            endpoints: &[DataEndpoint::Discovery],
2397            reliable: ReliableMode::Ordered,
2398            priority: 243,
2399            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2400        },
2401        DataTypeDefinition {
2402            id: DataType::ManagedVariableValue,
2403            name: "SEDSNET_MANAGED_VARIABLE_VALUE",
2404            description: "",
2405            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2406            endpoints: &[DataEndpoint::Discovery],
2407            reliable: ReliableMode::Ordered,
2408            priority: 243,
2409            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2410        },
2411        DataTypeDefinition {
2412            id: DataType::DiscoveryLeave,
2413            name: "SEDSNET_DISCOVERY_LEAVE",
2414            description: "",
2415            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2416            endpoints: &[DataEndpoint::Discovery],
2417            reliable: ReliableMode::None,
2418            priority: 244,
2419            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2420        },
2421        DataTypeDefinition {
2422            id: DataType::DiscoveryLinkCapabilities,
2423            name: "SEDSNET_DISCOVERY_LINK_CAPABILITIES",
2424            description: "",
2425            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2426            endpoints: &[DataEndpoint::Discovery],
2427            reliable: ReliableMode::None,
2428            priority: 240,
2429            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2430        },
2431        DataTypeDefinition {
2432            id: DataType::DiscoveryAddress,
2433            name: "SEDSNET_DISCOVERY_ADDRESS",
2434            description: "",
2435            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2436            endpoints: &[DataEndpoint::Discovery],
2437            reliable: ReliableMode::Ordered,
2438            priority: 244,
2439            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2440        },
2441        DataTypeDefinition {
2442            id: DataType::P2pMessage,
2443            name: "SEDSNET_P2P_MESSAGE",
2444            description: "",
2445            element: MessageElement::Dynamic(MessageDataType::UInt8, MessageClass::Data),
2446            endpoints: &[DataEndpoint::Discovery],
2447            reliable: ReliableMode::Ordered,
2448            priority: 246,
2449            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2450        },
2451    ]);
2452    types.extend_from_slice(EMBEDDED_SCHEMA_TYPES);
2453    types
2454}
2455
2456#[cfg(not(feature = "std"))]
2457pub fn merge_schema_snapshot(_snapshot: RuntimeSchemaSnapshot) -> SchemaMergeReport {
2458    SchemaMergeReport {
2459        endpoints_added: 0,
2460        endpoints_replaced: 0,
2461        endpoints_kept: 0,
2462        types_added: 0,
2463        types_replaced: 0,
2464        types_kept: 0,
2465    }
2466}
2467
2468#[cfg(not(feature = "std"))]
2469pub fn merge_owned_schema_snapshot_with_budget(
2470    _snapshot: OwnedRuntimeSchemaSnapshot,
2471    _max_schema_bytes: usize,
2472) -> TelemetryResult<SchemaMergeReport> {
2473    Ok(SchemaMergeReport {
2474        endpoints_added: 0,
2475        endpoints_replaced: 0,
2476        endpoints_kept: 0,
2477        types_added: 0,
2478        types_replaced: 0,
2479        types_kept: 0,
2480    })
2481}
2482
2483#[cfg(not(feature = "std"))]
2484pub fn schema_fingerprint() -> u64 {
2485    0
2486}
2487
2488#[cfg(not(feature = "std"))]
2489pub fn schema_bytes_used() -> usize {
2490    known_endpoints()
2491        .iter()
2492        .map(|def| {
2493            size_of::<EndpointDefinition>()
2494                .saturating_add(def.name.len())
2495                .saturating_add(def.description.len())
2496        })
2497        .sum::<usize>()
2498        .saturating_add(
2499            known_data_types()
2500                .iter()
2501                .map(|def| {
2502                    size_of::<DataTypeDefinition>()
2503                        .saturating_add(def.name.len())
2504                        .saturating_add(def.description.len())
2505                        .saturating_add(
2506                            def.endpoints
2507                                .len()
2508                                .saturating_mul(size_of::<DataEndpoint>()),
2509                        )
2510                })
2511                .sum::<usize>(),
2512        )
2513}
2514
2515#[cfg(not(feature = "std"))]
2516pub fn endpoint_exists(ep: DataEndpoint) -> bool {
2517    known_endpoints().iter().any(|def| def.id == ep)
2518}
2519
2520#[cfg(not(feature = "std"))]
2521pub fn data_type_exists(ty: DataType) -> bool {
2522    known_data_types().iter().any(|def| def.id == ty)
2523}
2524
2525#[cfg(not(feature = "std"))]
2526pub fn endpoint_definition(ep: DataEndpoint) -> Option<EndpointDefinition> {
2527    known_endpoints().into_iter().find(|def| def.id == ep)
2528}
2529
2530#[cfg(not(feature = "std"))]
2531pub fn data_type_definition(ty: DataType) -> Option<DataTypeDefinition> {
2532    known_data_types().into_iter().find(|def| def.id == ty)
2533}
2534
2535#[cfg(not(feature = "std"))]
2536pub fn endpoint_definition_by_name(name: &str) -> Option<EndpointDefinition> {
2537    known_endpoints().into_iter().find(|def| def.name == name)
2538}
2539
2540#[cfg(not(feature = "std"))]
2541pub fn data_type_definition_by_name(name: &str) -> Option<DataTypeDefinition> {
2542    known_data_types().into_iter().find(|def| def.name == name)
2543}
2544
2545#[cfg(not(feature = "std"))]
2546pub fn remove_endpoint(_ep: DataEndpoint) -> TelemetryResult<bool> {
2547    Err(TelemetryError::BadArg)
2548}
2549
2550#[cfg(not(feature = "std"))]
2551pub fn remove_endpoint_by_name(_name: &str) -> TelemetryResult<bool> {
2552    Err(TelemetryError::BadArg)
2553}
2554
2555#[cfg(not(feature = "std"))]
2556pub fn remove_data_type(_ty: DataType) -> TelemetryResult<bool> {
2557    Err(TelemetryError::BadArg)
2558}
2559
2560#[cfg(not(feature = "std"))]
2561pub fn remove_data_type_by_name(_name: &str) -> TelemetryResult<bool> {
2562    Err(TelemetryError::BadArg)
2563}
2564
2565#[cfg(not(feature = "std"))]
2566pub fn get_endpoint_meta(endpoint_type: DataEndpoint) -> EndpointMeta {
2567    known_endpoints()
2568        .iter()
2569        .find(|def| def.id == endpoint_type)
2570        .map(|def| EndpointMeta {
2571            name: def.name,
2572            description: def.description,
2573            link_local_only: def.link_local_only,
2574        })
2575        .unwrap_or(EndpointMeta {
2576            name: "UNKNOWN_ENDPOINT",
2577            description: "",
2578            link_local_only: false,
2579        })
2580}
2581
2582#[cfg(not(feature = "std"))]
2583pub fn get_message_meta(data_type: DataType) -> MessageMeta {
2584    known_data_types()
2585        .iter()
2586        .find(|def| def.id == data_type)
2587        .map(|def| MessageMeta {
2588            name: def.name,
2589            description: def.description,
2590            element: def.element,
2591            endpoints: def.endpoints,
2592            reliable: def.reliable,
2593            priority: def.priority,
2594            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2595        })
2596        .unwrap_or(MessageMeta {
2597            name: "UNKNOWN_TYPE",
2598            description: "",
2599            element: MessageElement::Dynamic(MessageDataType::Binary, MessageClass::Data),
2600            endpoints: &[],
2601            reliable: ReliableMode::None,
2602            priority: 0,
2603            e2e_encryption: E2eEncryptionPolicy::PreferOff,
2604        })
2605}
2606
2607#[cfg(not(feature = "std"))]
2608pub fn max_endpoint_id() -> u32 {
2609    known_endpoints()
2610        .iter()
2611        .map(|def| def.id.as_u32())
2612        .max()
2613        .unwrap_or(DataEndpoint::TelemetryError.as_u32())
2614}
2615
2616#[cfg(not(feature = "std"))]
2617pub fn max_data_type_id() -> u32 {
2618    known_data_types()
2619        .iter()
2620        .map(|def| def.id.as_u32())
2621        .max()
2622        .unwrap_or(DataType::DiscoverySchema.as_u32())
2623}
2624
2625// -----------------------------------------------------------------------------
2626// Optional JSON seeding for std builds
2627// -----------------------------------------------------------------------------
2628
2629#[cfg(feature = "std")]
2630pub fn register_schema_json_str(json: &str) -> TelemetryResult<()> {
2631    register_schema_json_bytes(json.as_bytes())
2632}
2633
2634#[cfg(feature = "std")]
2635pub fn register_schema_json_bytes(json: &[u8]) -> TelemetryResult<()> {
2636    register_schema_json_bytes_with_budget(json, MAX_QUEUE_BUDGET)
2637}
2638
2639/// Stream a schema document from an existing byte slice while limiting the
2640/// total retained schema memory. The caller owns the input slice; this function
2641/// never creates a second full-document allocation.
2642#[cfg(feature = "std")]
2643pub fn register_schema_json_bytes_with_budget(
2644    json: &[u8],
2645    max_schema_bytes: usize,
2646) -> TelemetryResult<()> {
2647    let max_input_bytes = schema_json_max_input_bytes(max_schema_bytes);
2648    if json.len() > max_input_bytes {
2649        return Err(TelemetryError::PacketTooLarge(
2650            "Schema JSON exceeds bounded input size",
2651        ));
2652    }
2653    let mut reg = registry().lock().expect("schema registry poisoned");
2654    register_schema_json_reader_into(&mut reg, json, false, max_schema_bytes)
2655}
2656
2657#[cfg(feature = "std")]
2658pub fn register_schema_json_file(path: impl AsRef<std::path::Path>) -> TelemetryResult<()> {
2659    register_schema_json_file_with_budget(path, MAX_QUEUE_BUDGET)
2660}
2661
2662/// Load a JSON schema using a fixed-size read buffer and a hard retained-memory
2663/// budget. Entries are decoded and validated individually, and a failed load is
2664/// rolled back completely.
2665#[cfg(feature = "std")]
2666pub fn register_schema_json_file_with_budget(
2667    path: impl AsRef<std::path::Path>,
2668    max_schema_bytes: usize,
2669) -> TelemetryResult<()> {
2670    let mut reg = registry().lock().expect("schema registry poisoned");
2671    register_schema_json_file_into(
2672        &mut reg,
2673        path.as_ref(),
2674        false,
2675        max_schema_bytes,
2676        SCHEMA_JSON_CHUNK_BYTES,
2677    )
2678}
2679
2680#[cfg(feature = "std")]
2681pub fn register_schema_json_path(path: &str) -> TelemetryResult<()> {
2682    register_schema_json_file(path)
2683}
2684
2685#[cfg(not(feature = "std"))]
2686pub fn register_schema_json_bytes(_json: &[u8]) -> TelemetryResult<()> {
2687    Err(TelemetryError::BadArg)
2688}
2689
2690#[cfg(feature = "serde")]
2691#[derive(serde::Deserialize)]
2692struct JsonConfig {
2693    endpoints: Vec<JsonEndpoint>,
2694    types: Vec<JsonType>,
2695}
2696
2697#[cfg(feature = "serde")]
2698#[derive(serde::Deserialize)]
2699struct JsonEndpoint {
2700    rust: Option<String>,
2701    name: String,
2702    #[serde(default, alias = "doc")]
2703    description: Option<String>,
2704    #[serde(default, alias = "link_local_only")]
2705    link_local_only: Option<bool>,
2706    #[serde(default, alias = "broadcast_mode")]
2707    broadcast_mode: Option<String>,
2708}
2709
2710#[cfg(feature = "serde")]
2711#[derive(serde::Deserialize)]
2712struct JsonType {
2713    rust: Option<String>,
2714    name: String,
2715    #[serde(default, alias = "doc")]
2716    description: Option<String>,
2717    class: String,
2718    element: JsonElement,
2719    endpoints: Vec<String>,
2720    #[serde(default)]
2721    reliable: Option<bool>,
2722    #[serde(default)]
2723    reliable_mode: Option<String>,
2724    #[serde(default)]
2725    priority: Option<u8>,
2726    #[serde(default)]
2727    e2e_encryption: Option<String>,
2728}
2729
2730fn parse_e2e_encryption_policy(raw: Option<&str>) -> TelemetryResult<E2eEncryptionPolicy> {
2731    match raw.unwrap_or("PreferOff") {
2732        "PreferOff" | "prefer_off" | "off" | "false" => Ok(E2eEncryptionPolicy::PreferOff),
2733        "PreferOn" | "prefer_on" | "preferred" | "true" => Ok(E2eEncryptionPolicy::PreferOn),
2734        "RequireOn" | "require_on" | "required" => Ok(E2eEncryptionPolicy::RequireOn),
2735        _ => Err(TelemetryError::BadArg),
2736    }
2737}
2738
2739#[cfg(feature = "serde")]
2740#[derive(serde::Deserialize)]
2741#[serde(tag = "kind")]
2742enum JsonElement {
2743    Static {
2744        data_type: String,
2745        count: Option<usize>,
2746    },
2747    Dynamic {
2748        data_type: String,
2749    },
2750}
2751
2752#[cfg(feature = "std")]
2753fn schema_json_max_input_bytes(max_schema_bytes: usize) -> usize {
2754    max_schema_bytes
2755        .saturating_mul(SCHEMA_JSON_INPUT_MULTIPLIER)
2756        .max(SCHEMA_JSON_CHUNK_BYTES)
2757}
2758
2759#[cfg(feature = "std")]
2760struct StreamingSchemaState<'a> {
2761    reg: &'a mut Registry,
2762    aliases: Vec<(String, DataEndpoint)>,
2763    alias_bytes: usize,
2764    added_endpoints: Vec<DataEndpoint>,
2765    added_types: Vec<DataType>,
2766    initial_next_endpoint_id: u32,
2767    initial_next_type_id: u32,
2768    link_local_overlay: bool,
2769    max_schema_bytes: usize,
2770    saw_endpoints: bool,
2771    saw_types: bool,
2772    error: Option<TelemetryError>,
2773}
2774
2775#[cfg(feature = "std")]
2776impl StreamingSchemaState<'_> {
2777    fn ensure_budget(&self, additional: usize) -> TelemetryResult<()> {
2778        if self
2779            .reg
2780            .schema_byte_cost()
2781            .saturating_add(self.alias_bytes)
2782            .saturating_add(additional)
2783            > self.max_schema_bytes
2784        {
2785            Err(TelemetryError::PacketTooLarge(
2786                "Schema exceeds maximum shared queue budget",
2787            ))
2788        } else {
2789            Ok(())
2790        }
2791    }
2792
2793    fn add_endpoint(&mut self, ep: JsonEndpoint) -> TelemetryResult<()> {
2794        let rust_name = ep.rust.unwrap_or_else(|| ep.name.clone());
2795        let id =
2796            known_endpoint_compat_id(&rust_name).unwrap_or(DataEndpoint(self.reg.next_endpoint_id));
2797        let existed = self.reg.endpoints.iter().any(|(known, _)| *known == id);
2798        let description = ep.description.unwrap_or_default();
2799        let retained = if existed {
2800            0
2801        } else {
2802            endpoint_schema_byte_cost(ep.name.len(), description.len())
2803        };
2804        self.ensure_budget(retained.saturating_add(rust_name.len()))?;
2805        self.reg.register_owned_endpoint(OwnedEndpointDefinition {
2806            id,
2807            name: ep.name,
2808            description,
2809            link_local_only: self.link_local_overlay
2810                || ep.link_local_only.unwrap_or(false)
2811                || matches!(ep.broadcast_mode.as_deref(), Some("Never")),
2812        })?;
2813        if !existed {
2814            self.added_endpoints.push(id);
2815        }
2816        self.alias_bytes = self.alias_bytes.saturating_add(rust_name.len());
2817        self.aliases.push((rust_name, id));
2818        Ok(())
2819    }
2820
2821    fn add_type(&mut self, ty: JsonType) -> TelemetryResult<()> {
2822        let rust_name = ty.rust.unwrap_or_else(|| ty.name.clone());
2823        let endpoints = ty
2824            .endpoints
2825            .iter()
2826            .map(|name| {
2827                self.aliases
2828                    .iter()
2829                    .find(|(alias, _)| alias == name)
2830                    .map(|(_, id)| *id)
2831                    .or_else(|| {
2832                        self.reg
2833                            .endpoints
2834                            .iter()
2835                            .find(|(_, meta)| meta.name.as_ref() == name.as_str())
2836                            .map(|(id, _)| *id)
2837                    })
2838                    .ok_or(TelemetryError::BadArg)
2839            })
2840            .collect::<TelemetryResult<Vec<_>>>()?;
2841        let id = known_type_compat_id(&rust_name).unwrap_or(DataType(self.reg.next_type_id));
2842        let existed = self.reg.types.iter().any(|(known, _)| *known == id);
2843        let description = ty.description.unwrap_or_default();
2844        if !existed {
2845            self.ensure_budget(type_schema_byte_cost(
2846                ty.name.len(),
2847                description.len(),
2848                endpoints.len(),
2849            ))?;
2850        }
2851        let class = parse_message_class(&ty.class)?;
2852        let element = match ty.element {
2853            JsonElement::Static { data_type, count } => MessageElement::Static(
2854                count.unwrap_or(1),
2855                parse_message_data_type(&data_type)?,
2856                class,
2857            ),
2858            JsonElement::Dynamic { data_type } => {
2859                MessageElement::Dynamic(parse_message_data_type(&data_type)?, class)
2860            }
2861        };
2862        let reliable = match ty.reliable_mode.as_deref() {
2863            Some("Ordered") => ReliableMode::Ordered,
2864            Some("Unordered") => ReliableMode::Unordered,
2865            Some("None") | None if ty.reliable.unwrap_or(false) => ReliableMode::Ordered,
2866            Some("None") | None => ReliableMode::None,
2867            _ => return Err(TelemetryError::BadArg),
2868        };
2869        self.reg.register_owned_type(OwnedDataTypeDefinition {
2870            id,
2871            name: ty.name,
2872            description,
2873            element,
2874            endpoints,
2875            reliable,
2876            priority: ty.priority.unwrap_or(0),
2877            e2e_encryption: parse_e2e_encryption_policy(ty.e2e_encryption.as_deref())?,
2878        })?;
2879        if !existed {
2880            self.added_types.push(id);
2881        }
2882        Ok(())
2883    }
2884
2885    fn rollback(&mut self) {
2886        self.reg
2887            .types
2888            .retain(|(id, _)| !self.added_types.contains(id));
2889        self.reg
2890            .endpoints
2891            .retain(|(id, _)| !self.added_endpoints.contains(id));
2892        self.reg.next_endpoint_id = self.initial_next_endpoint_id;
2893        self.reg.next_type_id = self.initial_next_type_id;
2894    }
2895}
2896
2897#[cfg(feature = "std")]
2898struct StreamingSchemaSeed<'a, 'r>(&'a mut StreamingSchemaState<'r>);
2899
2900#[cfg(feature = "std")]
2901impl<'de> serde::de::DeserializeSeed<'de> for StreamingSchemaSeed<'_, '_> {
2902    type Value = ();
2903
2904    fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
2905    where
2906        D: serde::Deserializer<'de>,
2907    {
2908        deserializer.deserialize_map(StreamingSchemaVisitor(self.0))
2909    }
2910}
2911
2912#[cfg(feature = "std")]
2913struct StreamingSchemaVisitor<'a, 'r>(&'a mut StreamingSchemaState<'r>);
2914
2915#[cfg(feature = "std")]
2916impl<'de> serde::de::Visitor<'de> for StreamingSchemaVisitor<'_, '_> {
2917    type Value = ();
2918
2919    fn expecting(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
2920        formatter.write_str("a schema object containing endpoints followed by types")
2921    }
2922
2923    fn visit_map<A>(self, mut map: A) -> Result<(), A::Error>
2924    where
2925        A: serde::de::MapAccess<'de>,
2926    {
2927        while let Some(key) = map.next_key::<String>()? {
2928            match key.as_str() {
2929                "endpoints" => {
2930                    self.0.saw_endpoints = true;
2931                    map.next_value_seed(JsonEndpointSequence(self.0))?;
2932                }
2933                "types" => {
2934                    if !self.0.saw_endpoints {
2935                        self.0.error = Some(TelemetryError::BadArg);
2936                        return Err(<A::Error as serde::de::Error>::custom(
2937                            "endpoints must precede types for bounded loading",
2938                        ));
2939                    }
2940                    self.0.saw_types = true;
2941                    map.next_value_seed(JsonTypeSequence(self.0))?;
2942                }
2943                _ => {
2944                    map.next_value::<serde::de::IgnoredAny>()?;
2945                }
2946            }
2947        }
2948        Ok(())
2949    }
2950}
2951
2952#[cfg(feature = "std")]
2953struct JsonEndpointSequence<'a, 'r>(&'a mut StreamingSchemaState<'r>);
2954
2955#[cfg(feature = "std")]
2956impl<'de> serde::de::DeserializeSeed<'de> for JsonEndpointSequence<'_, '_> {
2957    type Value = ();
2958
2959    fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
2960    where
2961        D: serde::Deserializer<'de>,
2962    {
2963        struct Visitor<'a, 'r>(&'a mut StreamingSchemaState<'r>);
2964        impl<'de> serde::de::Visitor<'de> for Visitor<'_, '_> {
2965            type Value = ();
2966            fn expecting(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
2967                f.write_str("an endpoint array")
2968            }
2969            fn visit_seq<A>(self, mut seq: A) -> Result<(), A::Error>
2970            where
2971                A: serde::de::SeqAccess<'de>,
2972            {
2973                while let Some(endpoint) = seq.next_element::<JsonEndpoint>()? {
2974                    if let Err(err) = self.0.add_endpoint(endpoint) {
2975                        self.0.error = Some(err);
2976                        return Err(<A::Error as serde::de::Error>::custom("schema endpoint"));
2977                    }
2978                }
2979                Ok(())
2980            }
2981        }
2982        deserializer.deserialize_seq(Visitor(self.0))
2983    }
2984}
2985
2986#[cfg(feature = "std")]
2987struct JsonTypeSequence<'a, 'r>(&'a mut StreamingSchemaState<'r>);
2988
2989#[cfg(feature = "std")]
2990impl<'de> serde::de::DeserializeSeed<'de> for JsonTypeSequence<'_, '_> {
2991    type Value = ();
2992
2993    fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
2994    where
2995        D: serde::Deserializer<'de>,
2996    {
2997        struct Visitor<'a, 'r>(&'a mut StreamingSchemaState<'r>);
2998        impl<'de> serde::de::Visitor<'de> for Visitor<'_, '_> {
2999            type Value = ();
3000            fn expecting(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
3001                f.write_str("a data type array")
3002            }
3003            fn visit_seq<A>(self, mut seq: A) -> Result<(), A::Error>
3004            where
3005                A: serde::de::SeqAccess<'de>,
3006            {
3007                while let Some(ty) = seq.next_element::<JsonType>()? {
3008                    if let Err(err) = self.0.add_type(ty) {
3009                        self.0.error = Some(err);
3010                        return Err(<A::Error as serde::de::Error>::custom("schema type"));
3011                    }
3012                }
3013                Ok(())
3014            }
3015        }
3016        deserializer.deserialize_seq(Visitor(self.0))
3017    }
3018}
3019
3020#[cfg(feature = "std")]
3021fn register_schema_json_reader_into<R: std::io::Read>(
3022    reg: &mut Registry,
3023    reader: R,
3024    link_local_overlay: bool,
3025    max_schema_bytes: usize,
3026) -> TelemetryResult<()> {
3027    let initial_next_endpoint_id = reg.next_endpoint_id;
3028    let initial_next_type_id = reg.next_type_id;
3029    let mut state = StreamingSchemaState {
3030        reg,
3031        aliases: Vec::new(),
3032        alias_bytes: 0,
3033        added_endpoints: Vec::new(),
3034        added_types: Vec::new(),
3035        initial_next_endpoint_id,
3036        initial_next_type_id,
3037        link_local_overlay,
3038        max_schema_bytes,
3039        saw_endpoints: false,
3040        saw_types: false,
3041        error: None,
3042    };
3043    let mut deserializer = serde_json::Deserializer::from_reader(reader);
3044    let parsed =
3045        serde::de::DeserializeSeed::deserialize(StreamingSchemaSeed(&mut state), &mut deserializer)
3046            .and_then(|()| deserializer.end());
3047    if parsed.is_err() || !state.saw_endpoints || !state.saw_types {
3048        let err = state
3049            .error
3050            .take()
3051            .unwrap_or(TelemetryError::Unpack("schema json"));
3052        state.rollback();
3053        return Err(err);
3054    }
3055    Ok(())
3056}
3057
3058#[cfg(feature = "std")]
3059fn register_schema_json_file_into(
3060    reg: &mut Registry,
3061    path: &std::path::Path,
3062    link_local_overlay: bool,
3063    max_schema_bytes: usize,
3064    chunk_bytes: usize,
3065) -> TelemetryResult<()> {
3066    let file = std::fs::File::open(path).map_err(|_| TelemetryError::Io("schema json file"))?;
3067    let max_input_bytes = schema_json_max_input_bytes(max_schema_bytes);
3068    if file
3069        .metadata()
3070        .map_err(|_| TelemetryError::Io("schema json metadata"))?
3071        .len()
3072        > max_input_bytes as u64
3073    {
3074        return Err(TelemetryError::PacketTooLarge(
3075            "Schema JSON exceeds bounded input size",
3076        ));
3077    }
3078    let bounded = std::io::Read::take(file, max_input_bytes.saturating_add(1) as u64);
3079    let reader = std::io::BufReader::with_capacity(chunk_bytes.max(1), bounded);
3080    register_schema_json_reader_into(reg, reader, link_local_overlay, max_schema_bytes)
3081}
3082
3083#[cfg(feature = "serde")]
3084fn json_config_to_snapshot(
3085    cfg: JsonConfig,
3086    link_local_overlay: bool,
3087    mut next_endpoint_id: u32,
3088    mut next_type_id: u32,
3089) -> TelemetryResult<OwnedRuntimeSchemaSnapshot> {
3090    let mut endpoint_ids: Vec<(String, DataEndpoint)> = Vec::new();
3091    let mut endpoints = Vec::with_capacity(cfg.endpoints.len());
3092    for ep in cfg.endpoints {
3093        let rust_name = ep.rust.clone().unwrap_or_else(|| ep.name.clone());
3094        let link_local = link_local_overlay
3095            || ep.link_local_only.unwrap_or(false)
3096            || matches!(ep.broadcast_mode.as_deref(), Some("Never"));
3097        let id = known_endpoint_compat_id(&rust_name).unwrap_or_else(|| {
3098            let id = DataEndpoint(next_endpoint_id);
3099            next_endpoint_id = next_endpoint_id.saturating_add(1);
3100            id
3101        });
3102        next_endpoint_id = next_endpoint_id.max(id.0.saturating_add(1));
3103        endpoints.push(OwnedEndpointDefinition {
3104            id,
3105            name: ep.name,
3106            description: ep.description.unwrap_or_default(),
3107            link_local_only: link_local,
3108        });
3109        endpoint_ids.push((rust_name, id));
3110    }
3111
3112    let mut types = Vec::with_capacity(cfg.types.len());
3113    for ty in cfg.types {
3114        let rust_name = ty.rust.clone().unwrap_or_else(|| ty.name.clone());
3115        let endpoints_for_type: Vec<DataEndpoint> = ty
3116            .endpoints
3117            .iter()
3118            .map(|name| {
3119                endpoint_ids
3120                    .iter()
3121                    .find(|(ep_name, _)| ep_name == name)
3122                    .map(|(_, id)| *id)
3123                    .ok_or(TelemetryError::BadArg)
3124            })
3125            .collect::<TelemetryResult<Vec<_>>>()?;
3126        let id = known_type_compat_id(&rust_name).unwrap_or_else(|| {
3127            let id = DataType(next_type_id);
3128            next_type_id = next_type_id.saturating_add(1);
3129            id
3130        });
3131        next_type_id = next_type_id.max(id.0.saturating_add(1));
3132        let class = parse_message_class(&ty.class)?;
3133        let element = match ty.element {
3134            JsonElement::Static { data_type, count } => MessageElement::Static(
3135                count.unwrap_or(1),
3136                parse_message_data_type(&data_type)?,
3137                class,
3138            ),
3139            JsonElement::Dynamic { data_type } => {
3140                MessageElement::Dynamic(parse_message_data_type(&data_type)?, class)
3141            }
3142        };
3143        let reliable = match ty.reliable_mode.as_deref() {
3144            Some("Ordered") => ReliableMode::Ordered,
3145            Some("Unordered") => ReliableMode::Unordered,
3146            Some("None") | None => {
3147                if ty.reliable.unwrap_or(false) {
3148                    ReliableMode::Ordered
3149                } else {
3150                    ReliableMode::None
3151                }
3152            }
3153            _ => return Err(TelemetryError::BadArg),
3154        };
3155        types.push(OwnedDataTypeDefinition {
3156            id,
3157            name: ty.name,
3158            description: ty.description.unwrap_or_default(),
3159            element,
3160            endpoints: endpoints_for_type,
3161            reliable,
3162            priority: ty.priority.unwrap_or(0),
3163            e2e_encryption: parse_e2e_encryption_policy(ty.e2e_encryption.as_deref())?,
3164        });
3165    }
3166    Ok(OwnedRuntimeSchemaSnapshot { endpoints, types })
3167}
3168
3169#[cfg(feature = "serde")]
3170pub fn schema_snapshot_from_json_bytes(json: &[u8]) -> TelemetryResult<OwnedRuntimeSchemaSnapshot> {
3171    let cfg: JsonConfig =
3172        serde_json::from_slice(json).map_err(|_| TelemetryError::Unpack("schema json"))?;
3173    json_config_to_snapshot(cfg, false, 100, 100)
3174}
3175
3176#[cfg(feature = "std")]
3177#[allow(dead_code)]
3178fn register_owned_schema_snapshot_into(
3179    reg: &mut Registry,
3180    snapshot: OwnedRuntimeSchemaSnapshot,
3181) -> TelemetryResult<()> {
3182    for endpoint in snapshot.endpoints {
3183        reg.register_owned_endpoint(endpoint)?;
3184    }
3185    for ty in snapshot.types {
3186        reg.register_owned_type(ty)?;
3187    }
3188    Ok(())
3189}
3190
3191#[cfg(feature = "serde")]
3192fn known_endpoint_compat_id(name: &str) -> Option<DataEndpoint> {
3193    match name {
3194        "SdCard" => Some(DataEndpoint(100)),
3195        "Radio" => Some(DataEndpoint(101)),
3196        "SoftwareBus" => Some(DataEndpoint(102)),
3197        _ => None,
3198    }
3199}
3200
3201#[cfg(feature = "serde")]
3202fn known_type_compat_id(name: &str) -> Option<DataType> {
3203    match name {
3204        "GpsData" => Some(DataType(100)),
3205        "ImuData" => Some(DataType(101)),
3206        "BatteryStatus" => Some(DataType(102)),
3207        "SystemStatus" => Some(DataType(103)),
3208        "BarometerData" => Some(DataType(104)),
3209        "MessageData" => Some(DataType(105)),
3210        "Heartbeat" => Some(DataType(106)),
3211        "IpcMessage" => Some(DataType(107)),
3212        _ => None,
3213    }
3214}
3215
3216#[cfg(feature = "serde")]
3217fn parse_message_class(s: &str) -> TelemetryResult<MessageClass> {
3218    match s {
3219        "Data" => Ok(MessageClass::Data),
3220        "Error" => Ok(MessageClass::Error),
3221        "Warning" => Ok(MessageClass::Warning),
3222        _ => Err(TelemetryError::BadArg),
3223    }
3224}
3225
3226#[cfg(feature = "serde")]
3227fn parse_message_data_type(s: &str) -> TelemetryResult<MessageDataType> {
3228    match s {
3229        "Float64" => Ok(MessageDataType::Float64),
3230        "Float32" => Ok(MessageDataType::Float32),
3231        "UInt8" => Ok(MessageDataType::UInt8),
3232        "UInt16" => Ok(MessageDataType::UInt16),
3233        "UInt32" => Ok(MessageDataType::UInt32),
3234        "UInt64" => Ok(MessageDataType::UInt64),
3235        "UInt128" => Ok(MessageDataType::UInt128),
3236        "Int8" => Ok(MessageDataType::Int8),
3237        "Int16" => Ok(MessageDataType::Int16),
3238        "Int32" => Ok(MessageDataType::Int32),
3239        "Int64" => Ok(MessageDataType::Int64),
3240        "Int128" => Ok(MessageDataType::Int128),
3241        "Bool" => Ok(MessageDataType::Bool),
3242        "String" => Ok(MessageDataType::String),
3243        "Binary" => Ok(MessageDataType::Binary),
3244        "NoData" => Ok(MessageDataType::NoData),
3245        _ => Err(TelemetryError::BadArg),
3246    }
3247}
3248
3249#[cfg(all(test, feature = "std"))]
3250pub(crate) fn seed_test_schema() {
3251    static SEEDED: OnceLock<()> = OnceLock::new();
3252    SEEDED.get_or_init(|| {
3253        let _ = register_schema_json_str(include_str!("../telemetry_config.test.json"));
3254        let ipc = include_str!("../telemetry_config.ipc.test.json");
3255        let mut reg = registry().lock().expect("schema registry poisoned");
3256        let _ = register_schema_json_reader_into(&mut reg, ipc.as_bytes(), true, MAX_QUEUE_BUDGET);
3257    });
3258}
3259
3260#[cfg(all(test, feature = "std"))]
3261mod memory_regression_tests {
3262    use super::*;
3263
3264    struct OneByteReader<'a> {
3265        remaining: &'a [u8],
3266    }
3267
3268    impl std::io::Read for OneByteReader<'_> {
3269        fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
3270            if out.is_empty() || self.remaining.is_empty() {
3271                return Ok(0);
3272            }
3273            out[0] = self.remaining[0];
3274            self.remaining = &self.remaining[1..];
3275            Ok(1)
3276        }
3277    }
3278
3279    #[test]
3280    fn runtime_schema_storage_is_owned_and_reclaimable() {
3281        let mut reg = Registry::new();
3282        let baseline_cost = reg.schema_byte_cost();
3283        let baseline_count = reg.endpoints.len();
3284
3285        for index in 0..256u32 {
3286            let id = DataEndpoint(50_000 + index);
3287            reg.register_owned_endpoint(OwnedEndpointDefinition {
3288                id,
3289                name: format!("MEMORY_RECLAIM_ENDPOINT_{index}"),
3290                description: "x".repeat(1024),
3291                link_local_only: false,
3292            })
3293            .unwrap();
3294            let name_handle = reg.endpoints.last().unwrap().1.name.clone();
3295            assert_eq!(Arc::strong_count(&name_handle), 2);
3296            drop(name_handle);
3297            reg.endpoints.retain(|(known, _)| *known != id);
3298        }
3299
3300        assert_eq!(reg.endpoints.len(), baseline_count);
3301        assert_eq!(reg.schema_byte_cost(), baseline_cost);
3302    }
3303
3304    #[test]
3305    fn streaming_schema_budget_failure_rolls_back_every_entry() {
3306        let mut reg = Registry::new();
3307        let baseline_cost = reg.schema_byte_cost();
3308        let baseline_endpoints = reg.endpoints.len();
3309        let baseline_types = reg.types.len();
3310        let next_endpoint_id = reg.next_endpoint_id;
3311        let next_type_id = reg.next_type_id;
3312        let json = br#"{
3313            "endpoints": [{
3314                "rust": "MemoryBudgetEndpoint",
3315                "name": "MEMORY_BUDGET_ENDPOINT",
3316                "description": "this entry must be rolled back"
3317            }],
3318            "types": []
3319        }"#;
3320
3321        let err = register_schema_json_reader_into(&mut reg, &json[..], false, baseline_cost)
3322            .expect_err("schema must exceed its unchanged baseline budget");
3323        assert!(matches!(err, TelemetryError::PacketTooLarge(_)));
3324        assert_eq!(reg.schema_byte_cost(), baseline_cost);
3325        assert_eq!(reg.endpoints.len(), baseline_endpoints);
3326        assert_eq!(reg.types.len(), baseline_types);
3327        assert_eq!(reg.next_endpoint_id, next_endpoint_id);
3328        assert_eq!(reg.next_type_id, next_type_id);
3329    }
3330
3331    #[test]
3332    fn streaming_schema_parse_failure_rolls_back_prior_chunks() {
3333        let mut reg = Registry::new();
3334        let baseline_cost = reg.schema_byte_cost();
3335        let baseline_endpoints = reg.endpoints.len();
3336        let next_endpoint_id = reg.next_endpoint_id;
3337        let json = br#"{
3338            "endpoints": [{
3339                "rust": "MemoryRollbackEndpoint",
3340                "name": "MEMORY_ROLLBACK_ENDPOINT"
3341            }],
3342            "types": [{
3343                "rust": "MemoryRollbackType",
3344                "name": "MEMORY_ROLLBACK_TYPE",
3345                "class": "NotAClass",
3346                "element": {"kind": "Dynamic", "data_type": "UInt8"},
3347                "endpoints": ["MemoryRollbackEndpoint"]
3348            }]
3349        }"#;
3350
3351        assert!(register_schema_json_reader_into(
3352            &mut reg,
3353            &json[..],
3354            false,
3355            baseline_cost + 16 * 1024,
3356        )
3357        .is_err());
3358        assert_eq!(reg.schema_byte_cost(), baseline_cost);
3359        assert_eq!(reg.endpoints.len(), baseline_endpoints);
3360        assert_eq!(reg.next_endpoint_id, next_endpoint_id);
3361    }
3362
3363    #[test]
3364    fn streaming_schema_loads_through_single_byte_reads() {
3365        let mut reg = Registry::new();
3366        let baseline_cost = reg.schema_byte_cost();
3367        let json = br#"{
3368            "endpoints": [{
3369                "rust": "SingleByteReaderEndpoint",
3370                "name": "SINGLE_BYTE_READER_ENDPOINT",
3371                "description": "streamed one byte at a time"
3372            }],
3373            "types": [{
3374                "rust": "SingleByteReaderType",
3375                "name": "SINGLE_BYTE_READER_TYPE",
3376                "class": "Data",
3377                "element": {"kind": "Dynamic", "data_type": "Binary"},
3378                "endpoints": ["SingleByteReaderEndpoint"]
3379            }]
3380        }"#;
3381
3382        register_schema_json_reader_into(
3383            &mut reg,
3384            OneByteReader { remaining: json },
3385            false,
3386            baseline_cost + 16 * 1024,
3387        )
3388        .unwrap();
3389
3390        assert!(
3391            reg.endpoints
3392                .iter()
3393                .any(|(_, meta)| meta.name.as_ref() == "SINGLE_BYTE_READER_ENDPOINT")
3394        );
3395        assert!(
3396            reg.types
3397                .iter()
3398                .any(|(_, meta)| meta.name.as_ref() == "SINGLE_BYTE_READER_TYPE")
3399        );
3400        assert!(reg.schema_byte_cost() > baseline_cost);
3401    }
3402}
sedsnet/config.rs

sedsnet/
config.rs
