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

sedsnet/
config.rs
