1#[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
33pub 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#[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#[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 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#[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#[derive(Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
433#[repr(transparent)]
434pub struct DataEndpoint(pub u32);
435
436impl DataEndpoint {
437 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 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#[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#[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 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
1818pub(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
2136fn 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#[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 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 let current = unsafe { &*node };
2830 if current.def.id == id
2831 && type_visible_at(¤t.def, EMBEDDED_TYPE_HEAD.load(Ordering::Acquire))
2832 {
2833 return Some(¤t.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(¤t.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 ¤t.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(¤t.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(¤t.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 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#[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#[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 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#[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#[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#[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}