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

sedsnet/
config.rs
