Skip to main content

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