1use std::convert::Infallible;
22
23use minicbor::data::Type;
24use minicbor::{Decoder, Encoder};
25use weida_core::Error;
26
27use crate::varint::{VarintError, decode_varint, encode_varint};
28
29pub mod limits {
33 pub const MAX_ENDPOINT_BYTES: usize = weida_core::MAX_PATH_BYTES;
36 pub const MAX_CONTENT_TYPE_BYTES: usize = 256;
38 pub const MAX_TRACEPARENT_BYTES: usize = weida_core::trace::MAX_TRACEPARENT_BYTES;
41 pub const MAX_TRACESTATE_BYTES: usize = weida_core::trace::MAX_TRACESTATE_BYTES;
44 pub const MAX_TOPIC_BYTES: usize = 256;
46 pub const PRODUCER_BYTES: usize = 32;
52 pub const MAX_FILTER_BYTES: usize = 256;
54 pub const MAX_MESSAGE_BYTES: usize = 1024;
56 pub const MAX_LIST_ITEMS: usize = 64;
61 pub const MAX_REPORT_LEVELS: usize = 16;
69 pub const MAX_SKIP_DEPTH: usize = 8;
71 pub const MAX_LAYER: u8 = 15;
74 pub const MAX_PATH_RECORD_BYTES: usize = 256;
76}
77
78mod hello_key {
80 pub const VERSIONS: u64 = 0;
81 pub const MAX_HEADER_BYTES: u64 = 1;
82 pub const MAX_TRANSFERS: u64 = 2;
83 pub const CAPABILITIES: u64 = 3;
84 pub const REQUIRED_CAPABILITIES: u64 = 4;
85 pub const GUARANTEES_OFFERED: u64 = 5;
86 pub const GUARANTEES_REQUIRED: u64 = 6;
87}
88
89mod data_key {
91 pub const ENDPOINT: u64 = 0;
92 pub const CONTENT_LEN: u64 = 1;
93 pub const CONTENT_TYPE: u64 = 2;
94 pub const TRACEPARENT: u64 = 3;
95 pub const TRACESTATE: u64 = 4;
96 pub const TOPIC: u64 = 5;
97 pub const SEQUENCE: u64 = 6;
98 pub const PRODUCER: u64 = 7;
99 pub const ACHIEVED: u64 = 8;
100 pub const REPORT_ID: u64 = 9;
101 pub const REPORT: u64 = 10;
102 pub const REPORT_MODE: u64 = 11;
103 pub const SEGMENT: u64 = 13;
106 pub const LAYER: u64 = 14;
107}
108
109mod error_key {
111 pub const CODE: u64 = 0;
112 pub const MESSAGE: u64 = 1;
113}
114
115mod subscription_key {
117 pub const ENDPOINT: u64 = 0;
118 pub const FILTER: u64 = 1;
119 pub const MAX_AGE_MS: u64 = 2;
120 pub const MAX_LAYER: u64 = 3;
121}
122
123mod credit_key {
125 pub const ENDPOINT: u64 = 0;
126 pub const FILTER: u64 = 1;
127 pub const LIMIT: u64 = 2;
128}
129
130mod cursor_key {
132 pub const REPORT_ID: u64 = 0;
133}
134
135mod flow_key {
137 pub const ENDPOINT: u64 = 0;
138 pub const FLOW: u64 = 1;
139 pub const CONTENT_TYPE: u64 = 2;
140 pub const TRACEPARENT: u64 = 3;
141 pub const TRACESTATE: u64 = 4;
142 pub const TOPIC: u64 = 5;
143}
144
145pub mod filter {
159 use super::HeaderError;
160
161 pub const SEPARATOR: char = '.';
163 pub const ONE_SEGMENT: &str = "*";
165 pub const REST: &str = "#";
167
168 pub fn matches(topic: &str, filter: &str) -> bool {
189 if filter.is_empty() {
190 return true;
191 }
192 let mut topic_segments = topic.split(SEPARATOR);
193 let mut filter_segments = filter.split(SEPARATOR);
194 loop {
195 let Some(pattern) = filter_segments.next() else {
196 return topic_segments.next().is_none();
198 };
199 if pattern == REST {
202 return true;
203 }
204 let Some(segment) = topic_segments.next() else {
205 return false;
206 };
207 if pattern != ONE_SEGMENT && pattern != segment {
208 return false;
209 }
210 }
211 }
212
213 pub fn validate(filter: &str) -> Result<(), HeaderError> {
220 if filter.len() > super::limits::MAX_FILTER_BYTES {
221 return Err(HeaderError::StringTooLong {
222 key: super::subscription_key::FILTER,
223 len: filter.len(),
224 max: super::limits::MAX_FILTER_BYTES,
225 });
226 }
227 let mut segments = filter.split(SEPARATOR).peekable();
228 while let Some(segment) = segments.next() {
229 let is_last = segments.peek().is_none();
230 if segment.contains(ONE_SEGMENT) && segment != ONE_SEGMENT {
231 return Err(HeaderError::InvalidFilter(
232 "`*` must occupy a whole segment",
233 ));
234 }
235 if segment.contains(REST) {
236 if segment != REST {
237 return Err(HeaderError::InvalidFilter(
238 "`#` must occupy a whole segment",
239 ));
240 }
241 if !is_last {
242 return Err(HeaderError::InvalidFilter("`#` must be the final segment"));
243 }
244 }
245 }
246 Ok(())
247 }
248}
249
250mod guarantee_key {
252 pub const DELIVERY: u64 = 0;
253 pub const ACKNOWLEDGEMENT: u64 = 1;
254 pub const DURABILITY: u64 = 2;
255 pub const REPLICAS: u64 = 3;
256 pub const ORDERING: u64 = 4;
257 pub const DEDUPLICATION: u64 = 5;
258 pub const DEDUP_WINDOW_MS: u64 = 6;
259 pub const BACKPRESSURE: u64 = 7;
260 pub const PRODUCER_NAMING: u64 = 8;
261 pub const CONTROL_ISOLATED: u64 = 9;
262}
263
264macro_rules! wire_enum {
275 ($(#[$meta:meta])* $name:ident { $($(#[$vmeta:meta])* $variant:ident = $value:literal),+ $(,)? }) => {
276 $(#[$meta])*
277 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
278 pub enum $name {
279 $($(#[$vmeta])* $variant,)+
280 }
281
282 impl $name {
283 pub fn to_wire(self) -> u64 {
285 match self {
286 $($name::$variant => $value,)+
287 }
288 }
289
290 pub fn from_wire(value: u64) -> Option<$name> {
293 match value {
294 $($value => Some($name::$variant),)+
295 _ => None,
296 }
297 }
298 }
299
300 const _: () = {
305 let values = [$($value as u64),+];
306 let mut i = 1;
307 while i < values.len() {
308 assert!(
309 values[i - 1] < values[i],
310 concat!(
311 stringify!($name),
312 ": wire values must ascend with declaration order, ",
313 "because the derived Ord is the ladder"
314 )
315 );
316 i += 1;
317 }
318 };
319 };
320}
321
322wire_enum! {
323 Delivery {
327 #[default]
329 BestEffort = 0,
330 AtMostOnce = 1,
332 AtLeastOnce = 2,
334 }
335}
336
337wire_enum! {
338 Acknowledgement {
341 None = 0,
343 #[default]
345 TransportReceipt = 1,
346 Accepted = 2,
348 Stored = 3,
350 Replicated = 4,
352 Processed = 5,
354 }
355}
356
357wire_enum! {
358 Durability {
361 #[default]
363 Written = 0,
364 Flushed = 1,
366 }
367}
368
369wire_enum! {
370 OrderingMode {
372 #[default]
374 None = 0,
375 PerProducerDetect = 1,
377 PerProducerReassemble = 2,
379 PerKey = 3,
381 Total = 4,
383 }
384}
385
386wire_enum! {
387 Deduplication {
389 #[default]
391 None = 0,
392 Bounded = 1,
394 Durable = 2,
396 }
397}
398
399wire_enum! {
400 Backpressure {
403 #[default]
405 Block = 0,
406 Reject = 1,
408 Drop = 2,
410 Spill = 3,
412 Coalesce = 4,
414 }
415}
416
417wire_enum! {
418 ProducerNaming {
422 #[default]
426 Fingerprint = 0,
427 Stable = 1,
429 }
430}
431
432wire_enum! {
433 ReportMode {
442 #[default]
445 Progress = 0,
446 FinalOnly = 1,
448 }
449}
450
451#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
462pub enum CursorLevel {
463 Known(Acknowledgement),
465 Application(u64),
467}
468
469impl CursorLevel {
470 pub const APPLICATION_FLOOR: u64 = 16;
472
473 pub fn to_wire(self) -> u64 {
475 match self {
476 CursorLevel::Known(level) => level.to_wire(),
477 CursorLevel::Application(value) => value,
478 }
479 }
480
481 pub fn from_wire(value: u64) -> Option<CursorLevel> {
484 if value >= CursorLevel::APPLICATION_FLOOR {
485 Some(CursorLevel::Application(value))
486 } else {
487 Acknowledgement::from_wire(value).map(CursorLevel::Known)
488 }
489 }
490
491 pub fn application(value: u64) -> Option<CursorLevel> {
494 (value >= CursorLevel::APPLICATION_FLOOR).then_some(CursorLevel::Application(value))
495 }
496}
497
498#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
507pub struct GuaranteeSet {
508 pub delivery: Delivery,
510 pub acknowledgement: Acknowledgement,
512 pub durability: Option<Durability>,
514 pub replicas: Option<u64>,
516 pub ordering: OrderingMode,
518 pub deduplication: Deduplication,
520 pub dedup_window_ms: Option<u64>,
522 pub backpressure: Backpressure,
524 pub producer_naming: ProducerNaming,
526 pub control_isolated: bool,
530}
531
532impl GuaranteeSet {
533 pub const CORE: GuaranteeSet = GuaranteeSet {
535 delivery: Delivery::BestEffort,
536 acknowledgement: Acknowledgement::TransportReceipt,
537 durability: None,
538 replicas: None,
539 ordering: OrderingMode::None,
540 deduplication: Deduplication::None,
541 dedup_window_ms: None,
542 backpressure: Backpressure::Block,
543 producer_naming: ProducerNaming::Fingerprint,
544 control_isolated: false,
545 };
546
547 pub fn is_core(&self) -> bool {
550 *self == GuaranteeSet::CORE
551 }
552
553 fn validate(&self) -> Result<(), HeaderError> {
555 let stored_or_replicated = matches!(
556 self.acknowledgement,
557 Acknowledgement::Stored | Acknowledgement::Replicated
558 );
559 if self.durability.is_some() && !stored_or_replicated {
560 return Err(HeaderError::InvalidGuarantees(
561 "durability without Stored or Replicated",
562 ));
563 }
564 match self.replicas {
565 Some(_) if self.acknowledgement != Acknowledgement::Replicated => {
566 return Err(HeaderError::InvalidGuarantees(
567 "replicas without Replicated",
568 ));
569 }
570 Some(n) if n < 2 => {
571 return Err(HeaderError::InvalidGuarantees(
572 "a replica count below 2 is not a replication",
573 ));
574 }
575 _ => {}
576 }
577 match (self.deduplication, self.dedup_window_ms) {
578 (Deduplication::Bounded, None) => {
579 return Err(HeaderError::InvalidGuarantees(
580 "Bounded deduplication without a window",
581 ));
582 }
583 (level, Some(_)) if level != Deduplication::Bounded => {
584 return Err(HeaderError::InvalidGuarantees(
585 "a dedup window without Bounded deduplication",
586 ));
587 }
588 _ => {}
589 }
590 Ok(())
591 }
592
593 fn encode_into(
595 &self,
596 e: &mut Encoder<Vec<u8>>,
597 ) -> Result<(), minicbor::encode::Error<Infallible>> {
598 let core = GuaranteeSet::CORE;
599 let count = u64::from(self.delivery != core.delivery)
600 + u64::from(self.acknowledgement != core.acknowledgement)
601 + u64::from(self.durability.is_some())
602 + u64::from(self.replicas.is_some())
603 + u64::from(self.ordering != core.ordering)
604 + u64::from(self.deduplication != core.deduplication)
605 + u64::from(self.dedup_window_ms.is_some())
606 + u64::from(self.backpressure != core.backpressure)
607 + u64::from(self.producer_naming != core.producer_naming)
608 + u64::from(self.control_isolated != core.control_isolated);
609 e.map(count)?;
610 if self.delivery != core.delivery {
611 e.u64(guarantee_key::DELIVERY)?
612 .u64(self.delivery.to_wire())?;
613 }
614 if self.acknowledgement != core.acknowledgement {
615 e.u64(guarantee_key::ACKNOWLEDGEMENT)?
616 .u64(self.acknowledgement.to_wire())?;
617 }
618 if let Some(durability) = self.durability {
619 e.u64(guarantee_key::DURABILITY)?
620 .u64(durability.to_wire())?;
621 }
622 if let Some(replicas) = self.replicas {
623 e.u64(guarantee_key::REPLICAS)?.u64(replicas)?;
624 }
625 if self.ordering != core.ordering {
626 e.u64(guarantee_key::ORDERING)?
627 .u64(self.ordering.to_wire())?;
628 }
629 if self.deduplication != core.deduplication {
630 e.u64(guarantee_key::DEDUPLICATION)?
631 .u64(self.deduplication.to_wire())?;
632 }
633 if let Some(window) = self.dedup_window_ms {
634 e.u64(guarantee_key::DEDUP_WINDOW_MS)?.u64(window)?;
635 }
636 if self.backpressure != core.backpressure {
637 e.u64(guarantee_key::BACKPRESSURE)?
638 .u64(self.backpressure.to_wire())?;
639 }
640 if self.producer_naming != core.producer_naming {
641 e.u64(guarantee_key::PRODUCER_NAMING)?
642 .u64(self.producer_naming.to_wire())?;
643 }
644 if self.control_isolated != core.control_isolated {
645 e.u64(guarantee_key::CONTROL_ISOLATED)?
646 .u64(u64::from(self.control_isolated))?;
647 }
648 Ok(())
649 }
650
651 fn decode_from(m: &mut MapReader<'_, '_>) -> Result<GuaranteeSet, HeaderError> {
656 let mut set = GuaranteeSet::CORE;
657 let mut inner = MapReader::new(m.d)?;
658 while let Some(key) = inner.next_key()? {
659 match key {
660 guarantee_key::DELIVERY => set.delivery = level(inner.u64()?, "delivery")?,
661 guarantee_key::ACKNOWLEDGEMENT => {
662 set.acknowledgement = level(inner.u64()?, "acknowledgement")?;
663 }
664 guarantee_key::DURABILITY => {
665 set.durability = Some(level(inner.u64()?, "durability")?);
666 }
667 guarantee_key::REPLICAS => set.replicas = Some(inner.u64()?),
668 guarantee_key::ORDERING => set.ordering = level(inner.u64()?, "ordering")?,
669 guarantee_key::DEDUPLICATION => {
670 set.deduplication = level(inner.u64()?, "deduplication")?;
671 }
672 guarantee_key::DEDUP_WINDOW_MS => set.dedup_window_ms = Some(inner.u64()?),
673 guarantee_key::BACKPRESSURE => {
674 set.backpressure = level(inner.u64()?, "backpressure")?;
675 }
676 guarantee_key::PRODUCER_NAMING => {
677 set.producer_naming = level(inner.u64()?, "producer naming")?;
678 }
679 guarantee_key::CONTROL_ISOLATED => {
680 set.control_isolated = match inner.u64()? {
681 0 => false,
682 1 => true,
683 _ => {
684 return Err(HeaderError::InvalidGuarantees(
685 "control_isolated is 0 or 1",
686 ));
687 }
688 };
689 }
690 _ => inner.skip()?,
691 }
692 }
693 set.validate()?;
694 Ok(set)
695 }
696
697 pub fn intersect(&self, other: &GuaranteeSet) -> Result<GuaranteeSet, &'static str> {
706 if self.backpressure != other.backpressure {
707 return Err("backpressure");
708 }
709 if self.producer_naming != other.producer_naming {
710 return Err("producer naming");
711 }
712 if self.durability.is_some()
713 && other.durability.is_some()
714 && self.durability != other.durability
715 {
716 return Err("durability");
717 }
718 if self.replicas.is_some() && other.replicas.is_some() && self.replicas != other.replicas {
719 return Err("replicas");
720 }
721
722 let acknowledgement = self.acknowledgement.min(other.acknowledgement);
723 let keeps_durability = matches!(
724 acknowledgement,
725 Acknowledgement::Stored | Acknowledgement::Replicated
726 );
727 let deduplication = self.deduplication.min(other.deduplication);
728 let mut merged = GuaranteeSet {
729 delivery: self.delivery.min(other.delivery),
730 acknowledgement,
731 durability: if keeps_durability {
740 self.durability.and(other.durability)
741 } else {
742 None
743 },
744 replicas: if acknowledgement == Acknowledgement::Replicated {
745 self.replicas.and(other.replicas)
746 } else {
747 None
748 },
749 ordering: self.ordering.min(other.ordering),
750 deduplication,
751 dedup_window_ms: None,
753 backpressure: self.backpressure,
754 producer_naming: self.producer_naming,
755 control_isolated: self.control_isolated && other.control_isolated,
756 };
757 if deduplication == Deduplication::Bounded {
758 merged.dedup_window_ms = match (self.dedup_window_ms, other.dedup_window_ms) {
759 (Some(a), Some(b)) => Some(a.min(b)),
760 (Some(a), None) | (None, Some(a)) => Some(a),
761 (None, None) => None,
762 };
763 }
764 Ok(merged)
765 }
766
767 pub fn reaches(&self, required: &GuaranteeSet) -> bool {
773 if self.delivery < required.delivery
774 || self.acknowledgement < required.acknowledgement
775 || self.ordering < required.ordering
776 || self.deduplication < required.deduplication
777 {
778 return false;
779 }
780 if self.backpressure != required.backpressure
781 || self.producer_naming != required.producer_naming
782 {
783 return false;
784 }
785 if !self.control_isolated && required.control_isolated {
786 return false;
787 }
788 match (self.durability, required.durability) {
789 (_, None) => {}
790 (Some(have), Some(want)) if have >= want => {}
791 _ => return false,
792 }
793 match (self.replicas, required.replicas) {
794 (_, None) => {}
795 (Some(have), Some(want)) if have >= want => {}
796 _ => return false,
797 }
798 match (self.dedup_window_ms, required.dedup_window_ms) {
799 (_, None) => {}
800 (Some(have), Some(want)) if have >= want => {}
801 _ => return false,
802 }
803 true
804 }
805}
806
807fn level<T: WireLevel>(value: u64, dimension: &'static str) -> Result<T, HeaderError> {
809 T::from_wire_value(value).ok_or(HeaderError::UnknownLevel { dimension, value })
810}
811
812fn layer(value: u64, reason: &'static str) -> Result<u8, HeaderError> {
815 u8::try_from(value)
816 .ok()
817 .filter(|layer| *layer <= limits::MAX_LAYER)
818 .ok_or(HeaderError::InvalidLayer(reason))
819}
820
821trait WireLevel: Sized {
823 fn from_wire_value(value: u64) -> Option<Self>;
824}
825
826macro_rules! impl_wire_level {
827 ($($name:ident),+ $(,)?) => {
828 $(impl WireLevel for $name {
829 fn from_wire_value(value: u64) -> Option<$name> {
830 $name::from_wire(value)
831 }
832 })+
833 };
834}
835
836impl_wire_level!(
837 Delivery,
838 Acknowledgement,
839 Durability,
840 OrderingMode,
841 Deduplication,
842 Backpressure,
843 ProducerNaming,
844 ReportMode,
845);
846
847#[derive(Clone, Debug, PartialEq, Eq)]
849pub enum HeaderError {
850 Malformed(&'static str),
852 Indefinite,
854 DuplicateKey(u64),
856 UnorderedKey(u64),
858 NonUintKey,
860 MissingKey(u64),
862 StringTooLong {
864 key: u64,
866 len: usize,
868 max: usize,
870 },
871 ListTooLong {
873 key: u64,
875 len: u64,
877 max: usize,
879 },
880 DepthExceeded,
882 TrailingBytes,
884 UnknownLevel {
887 dimension: &'static str,
889 value: u64,
891 },
892 InvalidGuarantees(&'static str),
895 InvalidFilter(&'static str),
897 InvalidReport(&'static str),
901 InvalidLayer(&'static str),
905 HeaderTooLong {
910 len: usize,
912 max: u64,
914 },
915 InvalidPathReport(&'static str),
918}
919
920impl std::fmt::Display for HeaderError {
921 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
922 match self {
923 HeaderError::Malformed(what) => write!(f, "malformed header: {what}"),
924 HeaderError::Indefinite => f.write_str("indefinite-length items are not allowed"),
925 HeaderError::DuplicateKey(k) => write!(f, "duplicate header key {k}"),
926 HeaderError::UnorderedKey(k) => {
927 write!(f, "header key {k} is out of ascending order")
928 }
929 HeaderError::NonUintKey => f.write_str("header key is not an unsigned integer"),
930 HeaderError::MissingKey(k) => write!(f, "required header key {k} is missing"),
931 HeaderError::StringTooLong { key, len, max } => {
932 write!(
933 f,
934 "key {key}: text of {len} bytes exceeds the {max} byte cap"
935 )
936 }
937 HeaderError::ListTooLong { key, len, max } => {
938 write!(
939 f,
940 "key {key}: list of {len} items exceeds the {max} item cap"
941 )
942 }
943 HeaderError::DepthExceeded => f.write_str("unknown field nested too deeply"),
944 HeaderError::TrailingBytes => f.write_str("trailing bytes after the header"),
945 HeaderError::UnknownLevel { dimension, value } => {
946 write!(f, "unknown {dimension} level {value}")
947 }
948 HeaderError::InvalidGuarantees(why) => write!(f, "invalid guarantee set: {why}"),
949 HeaderError::InvalidFilter(why) => write!(f, "invalid topic filter: {why}"),
950 HeaderError::InvalidReport(reason) => write!(f, "invalid report: {reason}"),
951 HeaderError::InvalidLayer(reason) => write!(f, "invalid layer: {reason}"),
952 HeaderError::InvalidPathReport(reason) => write!(f, "invalid path report: {reason}"),
953 HeaderError::HeaderTooLong { len, max } => {
954 write!(
955 f,
956 "header of {len} bytes exceeds the peer's {max} byte limit"
957 )
958 }
959 }
960 }
961}
962
963impl std::error::Error for HeaderError {}
964
965impl From<HeaderError> for Error {
966 fn from(e: HeaderError) -> Error {
967 Error::Protocol(e.to_string())
968 }
969}
970
971fn check_text(key: u64, value: Option<&str>, max: usize) -> Result<(), HeaderError> {
973 match value {
974 Some(s) if s.len() > max => Err(HeaderError::StringTooLong {
975 key,
976 len: s.len(),
977 max,
978 }),
979 _ => Ok(()),
980 }
981}
982
983fn encode_into_with(
990 out: &mut Vec<u8>,
991 f: impl FnOnce(&mut Encoder<Vec<u8>>) -> Result<(), minicbor::encode::Error<Infallible>>,
992) {
993 let mut e = Encoder::new(std::mem::take(out));
997 f(&mut e).expect("encoding into a Vec is infallible");
998 *out = e.into_writer();
999}
1000
1001pub fn skip_value(d: &mut Decoder<'_>, max_depth: usize) -> Result<(), HeaderError> {
1008 let mut stack: Vec<u64> = Vec::new();
1011 let mut remaining: u64 = 1;
1012
1013 loop {
1014 if remaining == 0 {
1015 match stack.pop() {
1016 Some(outer) => {
1017 remaining = outer;
1018 continue;
1019 }
1020 None => return Ok(()),
1021 }
1022 }
1023 remaining -= 1;
1024
1025 let ty = d
1026 .datatype()
1027 .map_err(|_| HeaderError::Malformed("truncated value"))?;
1028 let nested = match ty {
1029 Type::Bool => {
1030 d.bool().map_err(|_| HeaderError::Malformed("bool"))?;
1031 None
1032 }
1033 Type::Null => {
1034 d.null().map_err(|_| HeaderError::Malformed("null"))?;
1035 None
1036 }
1037 Type::Undefined => {
1038 d.undefined()
1039 .map_err(|_| HeaderError::Malformed("undefined"))?;
1040 None
1041 }
1042 Type::U8
1043 | Type::U16
1044 | Type::U32
1045 | Type::U64
1046 | Type::I8
1047 | Type::I16
1048 | Type::I32
1049 | Type::I64
1050 | Type::Int => {
1051 d.int().map_err(|_| HeaderError::Malformed("integer"))?;
1052 None
1053 }
1054 Type::F32 | Type::F64 => {
1055 d.f64().map_err(|_| HeaderError::Malformed("float"))?;
1056 None
1057 }
1058 Type::Bytes => {
1059 d.bytes()
1060 .map_err(|_| HeaderError::Malformed("byte string"))?;
1061 None
1062 }
1063 Type::String => {
1064 d.str().map_err(|_| HeaderError::Malformed("text string"))?;
1065 None
1066 }
1067 Type::Array => Some(
1068 d.array()
1069 .map_err(|_| HeaderError::Malformed("array"))?
1070 .ok_or(HeaderError::Indefinite)?,
1071 ),
1072 Type::Map => {
1073 let pairs = d
1074 .map()
1075 .map_err(|_| HeaderError::Malformed("map"))?
1076 .ok_or(HeaderError::Indefinite)?;
1077 Some(
1078 pairs
1079 .checked_mul(2)
1080 .ok_or(HeaderError::Malformed("map length overflow"))?,
1081 )
1082 }
1083 Type::BytesIndef | Type::StringIndef | Type::ArrayIndef | Type::MapIndef => {
1084 return Err(HeaderError::Indefinite);
1085 }
1086 Type::Break => return Err(HeaderError::Malformed("unexpected break")),
1087 Type::Tag => return Err(HeaderError::Malformed("tags are not allowed")),
1090 Type::F16 => return Err(HeaderError::Malformed("half floats are not allowed")),
1091 Type::Simple => return Err(HeaderError::Malformed("simple values are not allowed")),
1092 Type::Unknown(_) => return Err(HeaderError::Malformed("unknown major type")),
1093 };
1094
1095 if let Some(count) = nested
1096 && count > 0
1097 {
1098 if stack.len() >= max_depth {
1099 return Err(HeaderError::DepthExceeded);
1100 }
1101 stack.push(remaining);
1102 remaining = count;
1103 }
1104 }
1105}
1106
1107struct MapReader<'a, 'b> {
1109 d: &'a mut Decoder<'b>,
1110 remaining: u64,
1111 seen: u64,
1114 last: Option<u64>,
1121}
1122
1123impl<'a, 'b> MapReader<'a, 'b> {
1124 fn new(d: &'a mut Decoder<'b>) -> Result<MapReader<'a, 'b>, HeaderError> {
1125 let len = d
1126 .map()
1127 .map_err(|_| HeaderError::Malformed("header is not a map"))?
1128 .ok_or(HeaderError::Indefinite)?;
1129 Ok(MapReader {
1130 d,
1131 remaining: len,
1132 seen: 0,
1133 last: None,
1134 })
1135 }
1136
1137 fn next_key(&mut self) -> Result<Option<u64>, HeaderError> {
1138 if self.remaining == 0 {
1139 return Ok(None);
1140 }
1141 self.remaining -= 1;
1142 match self.d.datatype() {
1143 Ok(Type::U8 | Type::U16 | Type::U32 | Type::U64) => {}
1144 Ok(_) => return Err(HeaderError::NonUintKey),
1145 Err(_) => return Err(HeaderError::Malformed("truncated key")),
1146 }
1147 let key = self.d.u64().map_err(|_| HeaderError::NonUintKey)?;
1148 if let Some(prev) = self.last {
1149 if key == prev {
1150 return Err(HeaderError::DuplicateKey(key));
1151 }
1152 if key < prev {
1153 return Err(HeaderError::UnorderedKey(key));
1154 }
1155 }
1156 self.last = Some(key);
1157 if key < 64 {
1158 self.seen |= 1u64 << key;
1159 }
1160 Ok(Some(key))
1161 }
1162
1163 fn saw(&self, key: u64) -> bool {
1164 key < 64 && self.seen & (1u64 << key) != 0
1165 }
1166
1167 fn require(&self, key: u64) -> Result<(), HeaderError> {
1168 if self.saw(key) {
1169 Ok(())
1170 } else {
1171 Err(HeaderError::MissingKey(key))
1172 }
1173 }
1174
1175 fn u64(&mut self) -> Result<u64, HeaderError> {
1176 self.d
1177 .u64()
1178 .map_err(|_| HeaderError::Malformed("expected an unsigned integer"))
1179 }
1180
1181 fn text(&mut self, key: u64, max: usize) -> Result<String, HeaderError> {
1182 let s = self
1183 .d
1184 .str()
1185 .map_err(|_| HeaderError::Malformed("expected a text string"))?;
1186 if s.len() > max {
1187 return Err(HeaderError::StringTooLong {
1188 key,
1189 len: s.len(),
1190 max,
1191 });
1192 }
1193 Ok(s.to_owned())
1194 }
1195
1196 fn byte_array<const N: usize>(&mut self, key: u64) -> Result<[u8; N], HeaderError> {
1203 let bytes = self
1204 .d
1205 .bytes()
1206 .map_err(|_| HeaderError::Malformed("expected a byte string"))?;
1207 if bytes.len() > N {
1208 return Err(HeaderError::StringTooLong {
1209 key,
1210 len: bytes.len(),
1211 max: N,
1212 });
1213 }
1214 bytes
1215 .try_into()
1216 .map_err(|_| HeaderError::Malformed("byte string has the wrong length"))
1217 }
1218
1219 fn uint_list(&mut self, key: u64) -> Result<Vec<u64>, HeaderError> {
1220 let len = self
1221 .d
1222 .array()
1223 .map_err(|_| HeaderError::Malformed("expected an array"))?
1224 .ok_or(HeaderError::Indefinite)?;
1225 if len > limits::MAX_LIST_ITEMS as u64 {
1226 return Err(HeaderError::ListTooLong {
1227 key,
1228 len,
1229 max: limits::MAX_LIST_ITEMS,
1230 });
1231 }
1232 let mut out = Vec::with_capacity(len as usize);
1234 for _ in 0..len {
1235 out.push(self.u64()?);
1236 }
1237 Ok(out)
1238 }
1239
1240 fn report_levels(&mut self) -> Result<Vec<CursorLevel>, HeaderError> {
1248 let len = self
1249 .d
1250 .array()
1251 .map_err(|_| HeaderError::Malformed("expected an array"))?
1252 .ok_or(HeaderError::Indefinite)?;
1253 if len > limits::MAX_REPORT_LEVELS as u64 {
1254 return Err(HeaderError::InvalidReport("too many report levels"));
1255 }
1256 let mut out: Vec<CursorLevel> = Vec::with_capacity(len as usize);
1258 let mut last: Option<u64> = None;
1259 for _ in 0..len {
1260 let value = self.u64()?;
1261 if let Some(prev) = last
1262 && value <= prev
1263 {
1264 return Err(HeaderError::InvalidReport("report levels must ascend"));
1265 }
1266 last = Some(value);
1267 out.push(
1268 CursorLevel::from_wire(value).ok_or(HeaderError::UnknownLevel {
1269 dimension: "report",
1270 value,
1271 })?,
1272 );
1273 }
1274 Ok(out)
1275 }
1276
1277 fn skip(&mut self) -> Result<(), HeaderError> {
1278 skip_value(self.d, limits::MAX_SKIP_DEPTH)
1279 }
1280}
1281
1282fn finish(d: &Decoder<'_>) -> Result<(), HeaderError> {
1284 if d.position() == d.input().len() {
1285 Ok(())
1286 } else {
1287 Err(HeaderError::TrailingBytes)
1288 }
1289}
1290
1291#[derive(Clone, Debug, PartialEq, Eq)]
1293pub struct Hello {
1294 pub versions: Vec<u64>,
1296 pub max_header_bytes: u64,
1298 pub max_transfers: u64,
1300 pub capabilities: Vec<u64>,
1302 pub required_capabilities: Vec<u64>,
1304 pub guarantees_offered: Option<GuaranteeSet>,
1310 pub guarantees_required: Option<GuaranteeSet>,
1316}
1317
1318impl Hello {
1319 pub fn v0(max_header_bytes: u64, max_transfers: u64) -> Hello {
1322 Hello {
1323 versions: vec![crate::VERSION],
1324 max_header_bytes,
1325 max_transfers,
1326 capabilities: Vec::new(),
1327 required_capabilities: Vec::new(),
1328 guarantees_offered: None,
1329 guarantees_required: None,
1330 }
1331 }
1332
1333 pub fn offered(&self) -> GuaranteeSet {
1335 self.guarantees_offered.unwrap_or(GuaranteeSet::CORE)
1336 }
1337
1338 pub fn required(&self) -> GuaranteeSet {
1340 self.guarantees_required.unwrap_or(GuaranteeSet::CORE)
1341 }
1342
1343 pub fn encode(&self) -> Vec<u8> {
1345 let mut out = Vec::new();
1346 self.encode_into(&mut out);
1347 out
1348 }
1349
1350 pub fn encode_into(&self, out: &mut Vec<u8>) {
1353 encode_into_with(out, |e| {
1354 let offered = self.guarantees_offered.filter(|s| !s.is_core());
1357 let required = self.guarantees_required.filter(|s| !s.is_core());
1358 e.map(5 + u64::from(offered.is_some()) + u64::from(required.is_some()))?;
1359 e.u64(hello_key::VERSIONS)?
1360 .array(self.versions.len() as u64)?;
1361 for v in &self.versions {
1362 e.u64(*v)?;
1363 }
1364 e.u64(hello_key::MAX_HEADER_BYTES)?
1365 .u64(self.max_header_bytes)?;
1366 e.u64(hello_key::MAX_TRANSFERS)?.u64(self.max_transfers)?;
1367 e.u64(hello_key::CAPABILITIES)?
1368 .array(self.capabilities.len() as u64)?;
1369 for c in &self.capabilities {
1370 e.u64(*c)?;
1371 }
1372 e.u64(hello_key::REQUIRED_CAPABILITIES)?
1373 .array(self.required_capabilities.len() as u64)?;
1374 for c in &self.required_capabilities {
1375 e.u64(*c)?;
1376 }
1377 if let Some(set) = offered {
1378 e.u64(hello_key::GUARANTEES_OFFERED)?;
1379 set.encode_into(e)?;
1380 }
1381 if let Some(set) = required {
1382 e.u64(hello_key::GUARANTEES_REQUIRED)?;
1383 set.encode_into(e)?;
1384 }
1385 Ok(())
1386 })
1387 }
1388
1389 pub fn decode(bytes: &[u8]) -> Result<Hello, HeaderError> {
1391 let mut d = Decoder::new(bytes);
1392 let mut versions = Vec::new();
1393 let mut max_header_bytes = 0;
1394 let mut max_transfers = 0;
1395 let mut capabilities = Vec::new();
1396 let mut required_capabilities = Vec::new();
1397 let mut guarantees_offered = None;
1398 let mut guarantees_required = None;
1399 {
1400 let mut m = MapReader::new(&mut d)?;
1401 while let Some(key) = m.next_key()? {
1402 match key {
1403 hello_key::VERSIONS => versions = m.uint_list(key)?,
1404 hello_key::MAX_HEADER_BYTES => max_header_bytes = m.u64()?,
1405 hello_key::MAX_TRANSFERS => max_transfers = m.u64()?,
1406 hello_key::CAPABILITIES => capabilities = m.uint_list(key)?,
1407 hello_key::REQUIRED_CAPABILITIES => required_capabilities = m.uint_list(key)?,
1408 hello_key::GUARANTEES_OFFERED => {
1409 guarantees_offered = Some(GuaranteeSet::decode_from(&mut m)?);
1410 }
1411 hello_key::GUARANTEES_REQUIRED => {
1412 guarantees_required = Some(GuaranteeSet::decode_from(&mut m)?);
1413 }
1414 _ => m.skip()?,
1415 }
1416 }
1417 for key in [
1418 hello_key::VERSIONS,
1419 hello_key::MAX_HEADER_BYTES,
1420 hello_key::MAX_TRANSFERS,
1421 hello_key::CAPABILITIES,
1422 hello_key::REQUIRED_CAPABILITIES,
1423 ] {
1424 m.require(key)?;
1425 }
1426 }
1427 finish(&d)?;
1428 let hello = Hello {
1429 versions,
1430 max_header_bytes,
1431 max_transfers,
1432 capabilities,
1433 required_capabilities,
1434 guarantees_offered,
1435 guarantees_required,
1436 };
1437 if !hello.offered().reaches(&hello.required()) {
1440 return Err(HeaderError::InvalidGuarantees(
1441 "guarantees_required is not covered by guarantees_offered",
1442 ));
1443 }
1444 Ok(hello)
1445 }
1446}
1447
1448#[derive(Clone, Debug, Default, PartialEq, Eq)]
1455pub struct DataHeader {
1456 pub endpoint: Option<String>,
1458 pub content_len: Option<u64>,
1460 pub content_type: Option<String>,
1462 pub traceparent: Option<String>,
1464 pub tracestate: Option<String>,
1466 pub topic: Option<String>,
1470 pub sequence: Option<u64>,
1479 pub producer: Option<[u8; limits::PRODUCER_BYTES]>,
1490 pub achieved: Option<Acknowledgement>,
1512 pub report_id: Option<u64>,
1519 pub report: Vec<CursorLevel>,
1528 pub report_mode: ReportMode,
1532 pub segment: Option<u64>,
1541 pub layer: Option<u8>,
1547}
1548
1549impl DataHeader {
1550 pub fn addressed(endpoint: impl Into<String>) -> DataHeader {
1552 DataHeader {
1553 endpoint: Some(endpoint.into()),
1554 ..DataHeader::default()
1555 }
1556 }
1557
1558 pub fn reply() -> DataHeader {
1563 DataHeader::default()
1564 }
1565
1566 pub fn check(&self) -> Result<(), HeaderError> {
1575 check_text(
1576 data_key::ENDPOINT,
1577 self.endpoint.as_deref(),
1578 limits::MAX_ENDPOINT_BYTES,
1579 )?;
1580 check_text(
1581 data_key::CONTENT_TYPE,
1582 self.content_type.as_deref(),
1583 limits::MAX_CONTENT_TYPE_BYTES,
1584 )?;
1585 check_text(
1586 data_key::TRACEPARENT,
1587 self.traceparent.as_deref(),
1588 limits::MAX_TRACEPARENT_BYTES,
1589 )?;
1590 check_text(
1591 data_key::TRACESTATE,
1592 self.tracestate.as_deref(),
1593 limits::MAX_TRACESTATE_BYTES,
1594 )?;
1595 check_text(
1596 data_key::TOPIC,
1597 self.topic.as_deref(),
1598 limits::MAX_TOPIC_BYTES,
1599 )?;
1600 if let Some(layer) = self.layer {
1601 if layer > limits::MAX_LAYER {
1602 return Err(HeaderError::InvalidLayer("layer above 15"));
1603 }
1604 if self.segment.is_none() {
1605 return Err(HeaderError::InvalidLayer("layer without segment"));
1606 }
1607 }
1608 if !self.report.is_empty() {
1609 if self.report_id.is_none() {
1610 return Err(HeaderError::InvalidReport("report without report_id"));
1611 }
1612 let mut distinct: Vec<u64> = self.report.iter().map(|l| l.to_wire()).collect();
1614 distinct.sort_unstable();
1615 distinct.dedup();
1616 if distinct.len() > limits::MAX_REPORT_LEVELS {
1617 return Err(HeaderError::InvalidReport("too many report levels"));
1618 }
1619 } else if self.report_id.is_some() {
1620 return Err(HeaderError::InvalidReport("report_id without report"));
1621 }
1622 Ok(())
1623 }
1624
1625 pub fn encode(&self) -> Vec<u8> {
1642 let mut out = Vec::new();
1643 self.encode_into(&mut out);
1644 out
1645 }
1646
1647 pub fn encode_into(&self, out: &mut Vec<u8>) {
1650 let mut report: Vec<u64> = self.report.iter().map(|level| level.to_wire()).collect();
1653 report.sort_unstable();
1654 report.dedup();
1655 let count = u64::from(self.endpoint.is_some())
1656 + u64::from(self.content_len.is_some())
1657 + u64::from(self.content_type.is_some())
1658 + u64::from(self.traceparent.is_some())
1659 + u64::from(self.tracestate.is_some())
1660 + u64::from(self.topic.is_some())
1661 + u64::from(self.sequence.is_some())
1662 + u64::from(self.producer.is_some())
1663 + u64::from(self.achieved.is_some())
1664 + u64::from(self.report_id.is_some())
1665 + u64::from(!report.is_empty())
1666 + u64::from(self.report_mode != ReportMode::default())
1667 + u64::from(self.segment.is_some())
1668 + u64::from(self.layer.is_some());
1669 encode_into_with(out, |e| {
1670 e.map(count)?;
1671 if let Some(endpoint) = &self.endpoint {
1672 e.u64(data_key::ENDPOINT)?.str(endpoint)?;
1673 }
1674 if let Some(len) = self.content_len {
1675 e.u64(data_key::CONTENT_LEN)?.u64(len)?;
1676 }
1677 if let Some(ct) = &self.content_type {
1678 e.u64(data_key::CONTENT_TYPE)?.str(ct)?;
1679 }
1680 if let Some(tp) = &self.traceparent {
1681 e.u64(data_key::TRACEPARENT)?.str(tp)?;
1682 }
1683 if let Some(ts) = &self.tracestate {
1684 e.u64(data_key::TRACESTATE)?.str(ts)?;
1685 }
1686 if let Some(topic) = &self.topic {
1687 e.u64(data_key::TOPIC)?.str(topic)?;
1688 }
1689 if let Some(sequence) = self.sequence {
1692 e.u64(data_key::SEQUENCE)?.u64(sequence)?;
1693 }
1694 if let Some(producer) = &self.producer {
1695 e.u64(data_key::PRODUCER)?.bytes(producer)?;
1696 }
1697 if let Some(achieved) = self.achieved {
1698 e.u64(data_key::ACHIEVED)?.u64(achieved.to_wire())?;
1699 }
1700 if let Some(report_id) = self.report_id {
1701 e.u64(data_key::REPORT_ID)?.u64(report_id)?;
1702 }
1703 if !report.is_empty() {
1704 e.u64(data_key::REPORT)?.array(report.len() as u64)?;
1705 for value in &report {
1706 e.u64(*value)?;
1707 }
1708 }
1709 if self.report_mode != ReportMode::default() {
1713 e.u64(data_key::REPORT_MODE)?
1714 .u64(self.report_mode.to_wire())?;
1715 }
1716 if let Some(segment) = self.segment {
1717 e.u64(data_key::SEGMENT)?.u64(segment)?;
1718 }
1719 if let Some(layer) = self.layer {
1720 e.u64(data_key::LAYER)?.u64(u64::from(layer))?;
1721 }
1722 Ok(())
1723 })
1724 }
1725
1726 pub fn decode(bytes: &[u8]) -> Result<DataHeader, HeaderError> {
1728 let mut d = Decoder::new(bytes);
1729 let mut header = DataHeader::default();
1730 {
1731 let mut m = MapReader::new(&mut d)?;
1732 while let Some(key) = m.next_key()? {
1733 match key {
1734 data_key::ENDPOINT => {
1735 header.endpoint = Some(m.text(key, limits::MAX_ENDPOINT_BYTES)?)
1736 }
1737 data_key::CONTENT_LEN => header.content_len = Some(m.u64()?),
1738 data_key::CONTENT_TYPE => {
1739 header.content_type = Some(m.text(key, limits::MAX_CONTENT_TYPE_BYTES)?)
1740 }
1741 data_key::TRACEPARENT => {
1742 header.traceparent = Some(m.text(key, limits::MAX_TRACEPARENT_BYTES)?)
1743 }
1744 data_key::TRACESTATE => {
1745 header.tracestate = Some(m.text(key, limits::MAX_TRACESTATE_BYTES)?)
1746 }
1747 data_key::TOPIC => header.topic = Some(m.text(key, limits::MAX_TOPIC_BYTES)?),
1748 data_key::SEQUENCE => header.sequence = Some(m.u64()?),
1749 data_key::PRODUCER => header.producer = Some(m.byte_array(key)?),
1750 data_key::ACHIEVED => {
1755 let value = m.u64()?;
1756 header.achieved = Some(Acknowledgement::from_wire(value).ok_or(
1757 HeaderError::UnknownLevel {
1758 dimension: "achieved",
1759 value,
1760 },
1761 )?);
1762 }
1763 data_key::REPORT_ID => header.report_id = Some(m.u64()?),
1764 data_key::REPORT => header.report = m.report_levels()?,
1765 data_key::REPORT_MODE => {
1766 header.report_mode = level(m.u64()?, "report_mode")?;
1767 }
1768 data_key::SEGMENT => header.segment = Some(m.u64()?),
1769 data_key::LAYER => header.layer = Some(layer(m.u64()?, "layer above 15")?),
1770 _ => m.skip()?,
1771 }
1772 }
1773 }
1774 finish(&d)?;
1775 if header.layer.is_some() && header.segment.is_none() {
1778 return Err(HeaderError::InvalidLayer("layer without segment"));
1779 }
1780 if !header.report.is_empty() && header.report_id.is_none() {
1784 return Err(HeaderError::InvalidReport("report without report_id"));
1785 }
1786 if header.report_id.is_some() && header.report.is_empty() {
1787 return Err(HeaderError::InvalidReport("report_id without report"));
1788 }
1789 Ok(header)
1790 }
1791}
1792
1793#[derive(Clone, Debug, PartialEq, Eq)]
1798pub struct ErrorHeader {
1799 pub code: u64,
1801 pub message: Option<String>,
1803}
1804
1805impl ErrorHeader {
1806 pub fn new(code: weida_core::ErrorCode) -> ErrorHeader {
1808 ErrorHeader {
1809 code: code.to_wire(),
1810 message: None,
1811 }
1812 }
1813
1814 pub fn error_code(&self) -> Option<weida_core::ErrorCode> {
1816 weida_core::ErrorCode::from_wire(self.code)
1817 }
1818
1819 pub fn encode(&self) -> Vec<u8> {
1821 let mut out = Vec::new();
1822 self.encode_into(&mut out);
1823 out
1824 }
1825
1826 pub fn encode_into(&self, out: &mut Vec<u8>) {
1829 let count = 1 + u64::from(self.message.is_some());
1830 encode_into_with(out, |e| {
1831 e.map(count)?;
1832 e.u64(error_key::CODE)?.u64(self.code)?;
1833 if let Some(msg) = &self.message {
1834 e.u64(error_key::MESSAGE)?.str(msg)?;
1835 }
1836 Ok(())
1837 })
1838 }
1839
1840 pub fn decode(bytes: &[u8]) -> Result<ErrorHeader, HeaderError> {
1842 let mut d = Decoder::new(bytes);
1843 let mut code = 0;
1844 let mut message = None;
1845 {
1846 let mut m = MapReader::new(&mut d)?;
1847 while let Some(key) = m.next_key()? {
1848 match key {
1849 error_key::CODE => code = m.u64()?,
1850 error_key::MESSAGE => message = Some(m.text(key, limits::MAX_MESSAGE_BYTES)?),
1851 _ => m.skip()?,
1852 }
1853 }
1854 m.require(error_key::CODE)?;
1855 }
1856 finish(&d)?;
1857 Ok(ErrorHeader { code, message })
1858 }
1859}
1860
1861#[derive(Clone, Debug, PartialEq, Eq)]
1868pub struct SubscriptionHeader {
1869 pub endpoint: String,
1871 pub filter: String,
1874 pub max_age_ms: Option<u64>,
1879 pub max_layer: Option<u8>,
1884}
1885
1886impl SubscriptionHeader {
1887 pub fn new(endpoint: impl Into<String>, filter: impl Into<String>) -> SubscriptionHeader {
1889 SubscriptionHeader {
1890 endpoint: endpoint.into(),
1891 filter: filter.into(),
1892 max_age_ms: None,
1893 max_layer: None,
1894 }
1895 }
1896
1897 pub fn check(&self) -> Result<(), HeaderError> {
1900 check_text(
1901 subscription_key::ENDPOINT,
1902 Some(&self.endpoint),
1903 limits::MAX_ENDPOINT_BYTES,
1904 )?;
1905 filter::validate(&self.filter)?;
1906 if self
1907 .max_layer
1908 .is_some_and(|layer| layer > limits::MAX_LAYER)
1909 {
1910 return Err(HeaderError::InvalidLayer("max_layer above 15"));
1911 }
1912 Ok(())
1913 }
1914
1915 pub fn encode(&self) -> Vec<u8> {
1917 let mut out = Vec::new();
1918 self.encode_into(&mut out);
1919 out
1920 }
1921
1922 pub fn encode_into(&self, out: &mut Vec<u8>) {
1925 encode_into_with(out, |e| {
1930 e.map(2 + u64::from(self.max_age_ms.is_some()) + u64::from(self.max_layer.is_some()))?;
1931 e.u64(subscription_key::ENDPOINT)?.str(&self.endpoint)?;
1932 e.u64(subscription_key::FILTER)?.str(&self.filter)?;
1933 if let Some(max_age_ms) = self.max_age_ms {
1934 e.u64(subscription_key::MAX_AGE_MS)?.u64(max_age_ms)?;
1935 }
1936 if let Some(max_layer) = self.max_layer {
1937 e.u64(subscription_key::MAX_LAYER)?
1938 .u64(u64::from(max_layer))?;
1939 }
1940 Ok(())
1941 })
1942 }
1943
1944 pub fn decode(bytes: &[u8]) -> Result<SubscriptionHeader, HeaderError> {
1946 let mut d = Decoder::new(bytes);
1947 let mut endpoint = None;
1948 let mut filter = None;
1949 let mut max_age_ms = None;
1950 let mut max_layer = None;
1951 {
1952 let mut m = MapReader::new(&mut d)?;
1953 while let Some(key) = m.next_key()? {
1954 match key {
1955 subscription_key::ENDPOINT => {
1956 endpoint = Some(m.text(key, limits::MAX_ENDPOINT_BYTES)?)
1957 }
1958 subscription_key::FILTER => {
1959 filter = Some(m.text(key, limits::MAX_FILTER_BYTES)?)
1960 }
1961 subscription_key::MAX_AGE_MS => max_age_ms = Some(m.u64()?),
1962 subscription_key::MAX_LAYER => {
1963 max_layer = Some(layer(m.u64()?, "max_layer above 15")?);
1964 }
1965 _ => m.skip()?,
1966 }
1967 }
1968 m.require(subscription_key::ENDPOINT)?;
1969 m.require(subscription_key::FILTER)?;
1970 }
1971 let filter = filter.expect("presence checked above");
1976 filter::validate(&filter)?;
1977 finish(&d)?;
1978 Ok(SubscriptionHeader {
1979 endpoint: endpoint.expect("presence checked above"),
1980 filter,
1981 max_age_ms,
1982 max_layer,
1983 })
1984 }
1985}
1986
1987#[derive(Clone, Debug, PartialEq, Eq)]
1994pub struct FlowHeader {
1995 pub endpoint: String,
1997 pub flow: u64,
2000 pub content_type: Option<String>,
2002 pub traceparent: Option<String>,
2004 pub tracestate: Option<String>,
2006 pub topic: Option<String>,
2008}
2009
2010impl FlowHeader {
2011 pub fn new(endpoint: impl Into<String>, flow: u64) -> FlowHeader {
2013 FlowHeader {
2014 endpoint: endpoint.into(),
2015 flow,
2016 content_type: None,
2017 traceparent: None,
2018 tracestate: None,
2019 topic: None,
2020 }
2021 }
2022
2023 pub fn check(&self) -> Result<(), HeaderError> {
2026 check_text(
2027 flow_key::ENDPOINT,
2028 Some(&self.endpoint),
2029 limits::MAX_ENDPOINT_BYTES,
2030 )?;
2031 check_text(
2032 flow_key::CONTENT_TYPE,
2033 self.content_type.as_deref(),
2034 limits::MAX_CONTENT_TYPE_BYTES,
2035 )?;
2036 check_text(
2037 flow_key::TRACEPARENT,
2038 self.traceparent.as_deref(),
2039 limits::MAX_TRACEPARENT_BYTES,
2040 )?;
2041 check_text(
2042 flow_key::TRACESTATE,
2043 self.tracestate.as_deref(),
2044 limits::MAX_TRACESTATE_BYTES,
2045 )?;
2046 check_text(
2047 flow_key::TOPIC,
2048 self.topic.as_deref(),
2049 limits::MAX_TOPIC_BYTES,
2050 )
2051 }
2052
2053 pub fn encode(&self) -> Vec<u8> {
2055 let mut out = Vec::new();
2056 self.encode_into(&mut out);
2057 out
2058 }
2059
2060 pub fn encode_into(&self, out: &mut Vec<u8>) {
2063 let count = 2
2064 + u64::from(self.content_type.is_some())
2065 + u64::from(self.traceparent.is_some())
2066 + u64::from(self.tracestate.is_some())
2067 + u64::from(self.topic.is_some());
2068 encode_into_with(out, |e| {
2069 e.map(count)?;
2070 e.u64(flow_key::ENDPOINT)?.str(&self.endpoint)?;
2071 e.u64(flow_key::FLOW)?.u64(self.flow)?;
2072 if let Some(ct) = &self.content_type {
2073 e.u64(flow_key::CONTENT_TYPE)?.str(ct)?;
2074 }
2075 if let Some(tp) = &self.traceparent {
2076 e.u64(flow_key::TRACEPARENT)?.str(tp)?;
2077 }
2078 if let Some(ts) = &self.tracestate {
2079 e.u64(flow_key::TRACESTATE)?.str(ts)?;
2080 }
2081 if let Some(topic) = &self.topic {
2082 e.u64(flow_key::TOPIC)?.str(topic)?;
2083 }
2084 Ok(())
2085 })
2086 }
2087
2088 pub fn decode(bytes: &[u8]) -> Result<FlowHeader, HeaderError> {
2090 let mut d = Decoder::new(bytes);
2091 let mut endpoint = None;
2092 let mut flow = None;
2093 let mut content_type = None;
2094 let mut traceparent = None;
2095 let mut tracestate = None;
2096 let mut topic = None;
2097 {
2098 let mut m = MapReader::new(&mut d)?;
2099 while let Some(key) = m.next_key()? {
2100 match key {
2101 flow_key::ENDPOINT => endpoint = Some(m.text(key, limits::MAX_ENDPOINT_BYTES)?),
2102 flow_key::FLOW => flow = Some(m.u64()?),
2103 flow_key::CONTENT_TYPE => {
2104 content_type = Some(m.text(key, limits::MAX_CONTENT_TYPE_BYTES)?)
2105 }
2106 flow_key::TRACEPARENT => {
2107 traceparent = Some(m.text(key, limits::MAX_TRACEPARENT_BYTES)?)
2108 }
2109 flow_key::TRACESTATE => {
2110 tracestate = Some(m.text(key, limits::MAX_TRACESTATE_BYTES)?)
2111 }
2112 flow_key::TOPIC => topic = Some(m.text(key, limits::MAX_TOPIC_BYTES)?),
2113 _ => m.skip()?,
2114 }
2115 }
2116 m.require(flow_key::ENDPOINT)?;
2117 m.require(flow_key::FLOW)?;
2118 }
2119 finish(&d)?;
2120 Ok(FlowHeader {
2121 endpoint: endpoint.expect("presence checked above"),
2122 flow: flow.expect("presence checked above"),
2123 content_type,
2124 traceparent,
2125 tracestate,
2126 topic,
2127 })
2128 }
2129}
2130
2131#[derive(Clone, Debug, PartialEq, Eq)]
2155pub struct CreditHeader {
2156 pub endpoint: String,
2158 pub filter: String,
2161 pub limit: u64,
2164}
2165
2166impl CreditHeader {
2167 pub fn new(endpoint: impl Into<String>, filter: impl Into<String>, limit: u64) -> CreditHeader {
2169 CreditHeader {
2170 endpoint: endpoint.into(),
2171 filter: filter.into(),
2172 limit,
2173 }
2174 }
2175
2176 pub fn check(&self) -> Result<(), HeaderError> {
2178 check_text(
2179 credit_key::ENDPOINT,
2180 Some(&self.endpoint),
2181 limits::MAX_ENDPOINT_BYTES,
2182 )?;
2183 filter::validate(&self.filter)
2184 }
2185
2186 pub fn encode(&self) -> Vec<u8> {
2188 let mut out = Vec::new();
2189 self.encode_into(&mut out);
2190 out
2191 }
2192
2193 pub fn encode_into(&self, out: &mut Vec<u8>) {
2196 encode_into_with(out, |e| {
2197 e.map(3)?;
2198 e.u64(credit_key::ENDPOINT)?.str(&self.endpoint)?;
2199 e.u64(credit_key::FILTER)?.str(&self.filter)?;
2200 e.u64(credit_key::LIMIT)?.u64(self.limit)?;
2201 Ok(())
2202 })
2203 }
2204
2205 pub fn decode(bytes: &[u8]) -> Result<CreditHeader, HeaderError> {
2207 let mut d = Decoder::new(bytes);
2208 let mut endpoint = None;
2209 let mut filter = None;
2210 let mut limit = None;
2211 {
2212 let mut m = MapReader::new(&mut d)?;
2213 while let Some(key) = m.next_key()? {
2214 match key {
2215 credit_key::ENDPOINT => {
2216 endpoint = Some(m.text(key, limits::MAX_ENDPOINT_BYTES)?)
2217 }
2218 credit_key::FILTER => filter = Some(m.text(key, limits::MAX_FILTER_BYTES)?),
2219 credit_key::LIMIT => limit = Some(m.u64()?),
2220 _ => m.skip()?,
2221 }
2222 }
2223 m.require(credit_key::ENDPOINT)?;
2224 m.require(credit_key::FILTER)?;
2225 m.require(credit_key::LIMIT)?;
2226 }
2227 let filter = filter.expect("presence checked above");
2230 filter::validate(&filter)?;
2231 finish(&d)?;
2232 Ok(CreditHeader {
2233 endpoint: endpoint.expect("presence checked above"),
2234 filter,
2235 limit: limit.expect("presence checked above"),
2236 })
2237 }
2238}
2239
2240#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2248pub struct CursorHeader {
2249 pub report_id: u64,
2251}
2252
2253impl CursorHeader {
2254 pub fn encode(&self) -> Vec<u8> {
2256 let mut out = Vec::new();
2257 self.encode_into(&mut out);
2258 out
2259 }
2260
2261 pub fn encode_into(&self, out: &mut Vec<u8>) {
2264 encode_into_with(out, |e| {
2265 e.map(1)?;
2266 e.u64(cursor_key::REPORT_ID)?.u64(self.report_id)?;
2267 Ok(())
2268 })
2269 }
2270
2271 pub fn decode(bytes: &[u8]) -> Result<CursorHeader, HeaderError> {
2273 let mut d = Decoder::new(bytes);
2274 let mut report_id = None;
2275 {
2276 let mut m = MapReader::new(&mut d)?;
2277 while let Some(key) = m.next_key()? {
2278 match key {
2279 cursor_key::REPORT_ID => report_id = Some(m.u64()?),
2280 _ => m.skip()?,
2281 }
2282 }
2283 m.require(cursor_key::REPORT_ID)?;
2284 }
2285 finish(&d)?;
2286 Ok(CursorHeader {
2287 report_id: report_id.expect("presence checked above"),
2288 })
2289 }
2290}
2291
2292pub const MAX_CURSOR_RECORD_LEN: usize = 2 * crate::varint::MAX_ENCODED_LEN;
2294
2295pub fn encode_cursor_record(
2302 level: CursorLevel,
2303 offset: u64,
2304 out: &mut Vec<u8>,
2305) -> Result<(), VarintError> {
2306 encode_varint(level.to_wire(), out)?;
2307 encode_varint(offset, out)
2308}
2309
2310pub fn decode_cursor_record(
2319 input: &[u8],
2320) -> Result<Option<(CursorLevel, u64, usize)>, HeaderError> {
2321 let Ok((raw_level, level_len)) = decode_varint(input) else {
2324 return Ok(None);
2325 };
2326 let Ok((offset, offset_len)) = decode_varint(&input[level_len..]) else {
2327 return Ok(None);
2328 };
2329 let level = CursorLevel::from_wire(raw_level).ok_or(HeaderError::UnknownLevel {
2330 dimension: "cursor",
2331 value: raw_level,
2332 })?;
2333 Ok(Some((level, offset, level_len + offset_len)))
2334}
2335
2336#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
2339pub struct ReportHead;
2340
2341impl ReportHead {
2342 pub fn encode(&self) -> Vec<u8> {
2344 let mut out = Vec::new();
2345 encode_into_with(&mut out, |e| {
2346 e.map(0)?;
2347 Ok(())
2348 });
2349 out
2350 }
2351
2352 pub fn decode(bytes: &[u8]) -> Result<ReportHead, HeaderError> {
2354 let mut d = Decoder::new(bytes);
2355 {
2356 let mut m = MapReader::new(&mut d)?;
2357 while m.next_key()?.is_some() {
2358 m.skip()?;
2359 }
2360 }
2361 finish(&d)?;
2362 Ok(ReportHead)
2363 }
2364}
2365
2366mod path_key {
2368 pub const RTT_US: u64 = 0;
2369 pub const MIN_RTT_US: u64 = 1;
2370 pub const CWND: u64 = 2;
2371 pub const CONGESTION_EVENTS: u64 = 3;
2372 pub const LOST_PACKETS: u64 = 4;
2373 pub const LOST_BYTES: u64 = 5;
2374 pub const SENT_PACKETS: u64 = 6;
2375 pub const CURRENT_MTU: u64 = 7;
2376 pub const TX_DATAGRAMS: u64 = 8;
2377 pub const TX_BYTES: u64 = 9;
2378 pub const RX_DATAGRAMS: u64 = 10;
2379 pub const RX_BYTES: u64 = 11;
2380}
2381
2382#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
2388pub struct PathRecord {
2389 pub rtt_us: u64,
2391 pub min_rtt_us: u64,
2393 pub cwnd: u64,
2395 pub congestion_events: u64,
2397 pub lost_packets: u64,
2399 pub lost_bytes: u64,
2401 pub sent_packets: u64,
2403 pub current_mtu: u64,
2405 pub tx_datagrams: u64,
2407 pub tx_bytes: u64,
2409 pub rx_datagrams: u64,
2411 pub rx_bytes: u64,
2413}
2414
2415impl PathRecord {
2416 fn fields(&self) -> [(u64, u64); 12] {
2417 [
2418 (path_key::RTT_US, self.rtt_us),
2419 (path_key::MIN_RTT_US, self.min_rtt_us),
2420 (path_key::CWND, self.cwnd),
2421 (path_key::CONGESTION_EVENTS, self.congestion_events),
2422 (path_key::LOST_PACKETS, self.lost_packets),
2423 (path_key::LOST_BYTES, self.lost_bytes),
2424 (path_key::SENT_PACKETS, self.sent_packets),
2425 (path_key::CURRENT_MTU, self.current_mtu),
2426 (path_key::TX_DATAGRAMS, self.tx_datagrams),
2427 (path_key::TX_BYTES, self.tx_bytes),
2428 (path_key::RX_DATAGRAMS, self.rx_datagrams),
2429 (path_key::RX_BYTES, self.rx_bytes),
2430 ]
2431 }
2432
2433 pub fn encode_into(&self, out: &mut Vec<u8>) {
2436 let fields = self.fields();
2437 let present = fields.iter().filter(|(_, v)| *v != 0).count() as u64;
2438 let mut map = Vec::new();
2439 encode_into_with(&mut map, |e| {
2440 e.map(present)?;
2441 for (key, value) in fields {
2442 if value != 0 {
2443 e.u64(key)?.u64(value)?;
2444 }
2445 }
2446 Ok(())
2447 });
2448 encode_varint(map.len() as u64, out).expect("a record is far below 2^62 bytes");
2449 out.extend_from_slice(&map);
2450 }
2451
2452 pub fn decode(input: &[u8]) -> Result<Option<(PathRecord, usize)>, HeaderError> {
2460 let Ok((len, prefix)) = decode_varint(input) else {
2461 return Ok(None);
2462 };
2463 if len > limits::MAX_PATH_RECORD_BYTES as u64 {
2464 return Err(HeaderError::InvalidPathReport("record above 256 bytes"));
2465 }
2466 let len = len as usize;
2467 let Some(map) = input.get(prefix..prefix + len) else {
2468 return Ok(None);
2469 };
2470 let mut record = PathRecord::default();
2471 let mut d = Decoder::new(map);
2472 {
2473 let mut m = MapReader::new(&mut d)?;
2474 while let Some(key) = m.next_key()? {
2475 let slot = match key {
2476 path_key::RTT_US => &mut record.rtt_us,
2477 path_key::MIN_RTT_US => &mut record.min_rtt_us,
2478 path_key::CWND => &mut record.cwnd,
2479 path_key::CONGESTION_EVENTS => &mut record.congestion_events,
2480 path_key::LOST_PACKETS => &mut record.lost_packets,
2481 path_key::LOST_BYTES => &mut record.lost_bytes,
2482 path_key::SENT_PACKETS => &mut record.sent_packets,
2483 path_key::CURRENT_MTU => &mut record.current_mtu,
2484 path_key::TX_DATAGRAMS => &mut record.tx_datagrams,
2485 path_key::TX_BYTES => &mut record.tx_bytes,
2486 path_key::RX_DATAGRAMS => &mut record.rx_datagrams,
2487 path_key::RX_BYTES => &mut record.rx_bytes,
2488 _ => {
2489 m.skip()?;
2490 continue;
2491 }
2492 };
2493 *slot = m.u64()?;
2494 }
2495 }
2496 finish(&d)?;
2497 Ok(Some((record, prefix + len)))
2498 }
2499}
2500
2501#[cfg(test)]
2502mod tests {
2503 fn encode_with(
2506 f: impl FnOnce(
2507 &mut minicbor::Encoder<Vec<u8>>,
2508 ) -> Result<(), minicbor::encode::Error<std::convert::Infallible>>,
2509 ) -> Vec<u8> {
2510 let mut out = Vec::new();
2511 super::encode_into_with(&mut out, f);
2512 out
2513 }
2514
2515 use super::*;
2516 use weida_core::ErrorCode;
2517
2518 #[test]
2521 fn golden_data_request_header() {
2522 let h = DataHeader::addressed("/t");
2523 let bytes = h.encode();
2524 assert_eq!(bytes, vec![0xA1, 0x00, 0x62, 0x2F, 0x74]);
2525 assert_eq!(bytes.len(), 0x05);
2526 assert_eq!(DataHeader::decode(&bytes).unwrap(), h);
2527 }
2528
2529 #[test]
2530 fn golden_data_reply_header() {
2531 let h = DataHeader::reply();
2533 let bytes = h.encode();
2534 assert_eq!(bytes, vec![0xA0]);
2535 assert_eq!(bytes.len(), 0x01);
2536 assert_eq!(DataHeader::decode(&bytes).unwrap(), h);
2537 }
2538
2539 #[test]
2540 fn golden_hello_header() {
2541 let h = Hello::v0(16384, 1024);
2542 let bytes = h.encode();
2543 assert_eq!(
2544 bytes,
2545 vec![
2546 0xA5, 0x00, 0x81, 0x00, 0x01, 0x19, 0x40, 0x00, 0x02, 0x19, 0x04, 0x00, 0x03, 0x80,
2547 0x04, 0x80
2548 ]
2549 );
2550 assert_eq!(bytes.len(), 0x10);
2551 assert_eq!(Hello::decode(&bytes).unwrap(), h);
2552 }
2553
2554 #[test]
2555 fn golden_error_header() {
2556 let h = ErrorHeader::new(ErrorCode::NoReply);
2557 let bytes = h.encode();
2558 assert_eq!(bytes, vec![0xA1, 0x00, 0x05]);
2559 assert_eq!(bytes.len(), 0x03);
2560 assert_eq!(ErrorHeader::decode(&bytes).unwrap(), h);
2561 }
2562
2563 #[test]
2564 fn golden_pub_copy_data_header() {
2565 let mut h = DataHeader::addressed("/md");
2566 h.topic = Some("px.eur".into());
2567 let bytes = h.encode();
2568 assert_eq!(
2569 bytes,
2570 vec![
2571 0xA2, 0x00, 0x63, 0x2F, 0x6D, 0x64, 0x05, 0x66, 0x70, 0x78, 0x2E, 0x65, 0x75, 0x72
2572 ]
2573 );
2574 assert_eq!(bytes.len(), 0x0E);
2575 assert_eq!(DataHeader::decode(&bytes).unwrap(), h);
2576 }
2577
2578 const VECTOR_PRODUCER: [u8; limits::PRODUCER_BYTES] = [
2581 0x9F, 0x86, 0xD0, 0x81, 0x88, 0x4C, 0x7D, 0x65, 0x9A, 0x2F, 0xEA, 0xA0, 0xC5, 0x5A, 0xD0,
2582 0x15, 0xA3, 0xBF, 0x4F, 0x1B, 0x2B, 0x0B, 0x82, 0x2C, 0xD1, 0x5D, 0x6C, 0x15, 0xB0, 0xF0,
2583 0x0A, 0x08,
2584 ];
2585
2586 #[test]
2587 fn golden_sequenced_data_header() {
2588 let mut h = DataHeader::addressed("/t");
2589 h.sequence = Some(1);
2590 let bytes = h.encode();
2591 assert_eq!(bytes, vec![0xA2, 0x00, 0x62, 0x2F, 0x74, 0x06, 0x01]);
2592 assert_eq!(bytes.len(), 0x07);
2593 assert_eq!(DataHeader::decode(&bytes).unwrap(), h);
2594 }
2595
2596 #[test]
2597 fn golden_relayed_data_header() {
2598 let mut h = DataHeader::addressed("/t");
2599 h.sequence = Some(1);
2600 h.producer = Some(VECTOR_PRODUCER);
2601 let bytes = h.encode();
2602 let mut expected = vec![0xA3, 0x00, 0x62, 0x2F, 0x74, 0x06, 0x01, 0x07, 0x58, 0x20];
2603 expected.extend_from_slice(&VECTOR_PRODUCER);
2604 assert_eq!(bytes, expected);
2605 assert_eq!(bytes.len(), 0x2A);
2606 assert_eq!(DataHeader::decode(&bytes).unwrap(), h);
2607 }
2608
2609 #[test]
2610 fn a_producer_longer_than_the_cap_is_rejected() {
2611 let bytes = encode_with(|e| {
2612 e.map(1)?;
2613 e.u64(data_key::PRODUCER)?
2614 .bytes(&[0u8; limits::PRODUCER_BYTES + 1])?;
2615 Ok(())
2616 });
2617 assert_eq!(
2618 DataHeader::decode(&bytes),
2619 Err(HeaderError::StringTooLong {
2620 key: data_key::PRODUCER,
2621 len: limits::PRODUCER_BYTES + 1,
2622 max: limits::PRODUCER_BYTES,
2623 })
2624 );
2625 }
2626
2627 #[test]
2628 fn a_producer_shorter_than_a_digest_is_rejected() {
2629 let bytes = encode_with(|e| {
2632 e.map(1)?;
2633 e.u64(data_key::PRODUCER)?.bytes(&[0u8; 16])?;
2634 Ok(())
2635 });
2636 assert!(matches!(
2637 DataHeader::decode(&bytes),
2638 Err(HeaderError::Malformed(_))
2639 ));
2640 }
2641
2642 #[test]
2643 fn the_new_keys_reject_the_wrong_cbor_type() {
2644 let sequence_as_text = encode_with(|e| {
2645 e.map(1)?;
2646 e.u64(data_key::SEQUENCE)?.str("7")?;
2647 Ok(())
2648 });
2649 assert!(DataHeader::decode(&sequence_as_text).is_err());
2650
2651 let producer_as_text = encode_with(|e| {
2652 e.map(1)?;
2653 e.u64(data_key::PRODUCER)?.str("sha256:…")?;
2654 Ok(())
2655 });
2656 assert!(DataHeader::decode(&producer_as_text).is_err());
2657 }
2658
2659 #[test]
2660 fn a_v0_header_carries_neither_new_key() {
2661 let mut h = DataHeader::addressed("/t");
2665 h.traceparent = Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01".into());
2666 let bytes = h.encode();
2667 let mut d = Decoder::new(&bytes);
2668 let pairs = d.map().unwrap().unwrap();
2669 let keys: Vec<u64> = (0..pairs)
2670 .map(|_| {
2671 let key = d.u64().unwrap();
2672 d.skip().unwrap();
2673 key
2674 })
2675 .collect();
2676 assert_eq!(keys, vec![data_key::ENDPOINT, data_key::TRACEPARENT]);
2677 }
2678
2679 #[test]
2680 fn golden_subscription_headers() {
2681 let h = SubscriptionHeader::new("/md", "px.");
2682 let bytes = h.encode();
2683 assert_eq!(
2684 bytes,
2685 vec![
2686 0xA2, 0x00, 0x63, 0x2F, 0x6D, 0x64, 0x01, 0x63, 0x70, 0x78, 0x2E
2687 ]
2688 );
2689 assert_eq!(bytes.len(), 0x0B);
2690 assert_eq!(SubscriptionHeader::decode(&bytes).unwrap(), h);
2693 }
2694
2695 #[test]
2698 fn data_header_roundtrip_with_every_field() {
2699 let h = DataHeader {
2700 endpoint: Some("/transform".into()),
2701 content_len: Some(1 << 40),
2702 content_type: Some("application/octet-stream".into()),
2703 traceparent: Some("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01".into()),
2704 tracestate: Some("vendor=value".into()),
2705 topic: Some("px.eur".into()),
2706 sequence: Some(u64::MAX),
2707 producer: Some([0x5A; limits::PRODUCER_BYTES]),
2708 achieved: Some(Acknowledgement::Processed),
2709 report_id: Some(7),
2710 report: vec![
2711 CursorLevel::Known(Acknowledgement::Accepted),
2712 CursorLevel::Known(Acknowledgement::Processed),
2713 CursorLevel::Application(CursorLevel::APPLICATION_FLOOR),
2714 ],
2715 report_mode: ReportMode::FinalOnly,
2716 segment: Some(u64::MAX),
2717 layer: Some(limits::MAX_LAYER),
2718 };
2719 assert_eq!(DataHeader::decode(&h.encode()).unwrap(), h);
2720 }
2721
2722 #[test]
2723 fn error_header_roundtrip_with_and_without_message() {
2724 let bare = ErrorHeader::new(ErrorCode::UnknownEndpoint);
2725 assert_eq!(ErrorHeader::decode(&bare.encode()).unwrap(), bare);
2726 assert_eq!(bare.error_code(), Some(ErrorCode::UnknownEndpoint));
2727
2728 let with_msg = ErrorHeader {
2729 code: 4,
2730 message: Some("handler panicked".into()),
2731 };
2732 assert_eq!(ErrorHeader::decode(&with_msg.encode()).unwrap(), with_msg);
2733 }
2734
2735 #[test]
2736 fn keys_are_emitted_in_ascending_order() {
2737 let h = DataHeader {
2738 endpoint: Some("/x".into()),
2739 content_len: Some(1),
2740 content_type: Some("t".into()),
2741 traceparent: Some("p".into()),
2742 tracestate: Some("s".into()),
2743 topic: Some("k".into()),
2744 sequence: Some(9),
2745 producer: Some([0u8; limits::PRODUCER_BYTES]),
2746 achieved: Some(Acknowledgement::Accepted),
2747 report_id: Some(1),
2748 report: vec![CursorLevel::Known(Acknowledgement::Stored)],
2749 report_mode: ReportMode::FinalOnly,
2750 segment: Some(3),
2751 layer: Some(2),
2752 };
2753 let bytes = h.encode();
2754 let mut d = Decoder::new(&bytes);
2755 let n = d.map().unwrap().unwrap();
2756 let mut last = None;
2757 for _ in 0..n {
2758 let key = d.u64().unwrap();
2759 if let Some(prev) = last {
2760 assert!(key > prev, "keys must ascend: {prev} then {key}");
2761 }
2762 last = Some(key);
2763 d.skip().unwrap();
2764 }
2765 }
2766
2767 #[test]
2768 fn an_unsorted_report_with_repeats_is_emitted_as_the_canonical_ascending_set() {
2769 let h = DataHeader {
2774 report_id: Some(1),
2775 report: vec![
2776 CursorLevel::Application(CursorLevel::APPLICATION_FLOOR),
2777 CursorLevel::Known(Acknowledgement::Stored),
2778 CursorLevel::Application(CursorLevel::APPLICATION_FLOOR),
2779 CursorLevel::Known(Acknowledgement::Accepted),
2780 CursorLevel::Known(Acknowledgement::Stored),
2781 ],
2782 ..DataHeader::reply()
2783 };
2784 let decoded = DataHeader::decode(&h.encode()).expect("encoder emits the canonical form");
2785 assert_eq!(
2786 decoded.report,
2787 vec![
2788 CursorLevel::Known(Acknowledgement::Accepted),
2789 CursorLevel::Known(Acknowledgement::Stored),
2790 CursorLevel::Application(CursorLevel::APPLICATION_FLOOR),
2791 ]
2792 );
2793 }
2794
2795 #[test]
2798 fn every_data_field_is_optional_at_the_decoder() {
2799 assert_eq!(DataHeader::decode(&[0xA0]).unwrap(), DataHeader::default());
2803
2804 let only_topic = encode_with(|e| {
2805 e.map(1)?;
2806 e.u64(data_key::TOPIC)?.str("px.eur")?;
2807 Ok(())
2808 });
2809 let h = DataHeader::decode(&only_topic).unwrap();
2810 assert_eq!(h.topic.as_deref(), Some("px.eur"));
2811 assert_eq!(h.endpoint, None);
2812 }
2813
2814 #[test]
2815 fn absent_fields_are_omitted_by_the_encoder() {
2816 let h = DataHeader::addressed("/t");
2817 assert_eq!(h.encode(), vec![0xA1, 0x00, 0x62, 0x2F, 0x74]);
2818 }
2819
2820 #[test]
2823 fn unknown_keys_are_skipped() {
2824 let h = DataHeader::addressed("/t");
2827 let extended = encode_with(|e| {
2828 e.map(2)?;
2829 e.u64(0)?.str("/t")?;
2830 e.u64(63)?.array(2)?.u64(7)?.map(1)?.u64(1)?.bool(true)?;
2831 Ok(())
2832 });
2833 assert_eq!(DataHeader::decode(&extended).unwrap(), h);
2834 }
2835
2836 #[test]
2837 fn unknown_keys_above_the_reserved_range_are_skipped() {
2838 let extended = encode_with(|e| {
2839 e.map(2)?;
2840 e.u64(1)?.u64(5)?;
2841 e.u64(1000)?.str("future")?;
2842 Ok(())
2843 });
2844 let h = DataHeader::decode(&extended).unwrap();
2845 assert_eq!(h.content_len, Some(5));
2846 }
2847
2848 #[test]
2849 fn skipping_tolerates_nesting_up_to_the_depth_limit() {
2850 for depth in [1usize, limits::MAX_SKIP_DEPTH] {
2851 let bytes = encode_with(|e| {
2852 e.map(2)?;
2853 e.u64(data_key::CONTENT_LEN)?.u64(1)?;
2854 e.u64(50)?;
2855 for _ in 0..depth {
2856 e.array(1)?;
2857 }
2858 e.u64(1)?;
2859 Ok(())
2860 });
2861 let h = DataHeader::decode(&bytes).unwrap_or_else(|e| panic!("depth {depth}: {e}"));
2862 assert_eq!(h.content_len, Some(1), "depth {depth}");
2863 }
2864 }
2865
2866 #[test]
2867 fn skipping_rejects_nesting_beyond_the_depth_limit() {
2868 let bytes = encode_with(|e| {
2869 e.map(1)?;
2870 e.u64(50)?;
2871 for _ in 0..(limits::MAX_SKIP_DEPTH + 1) {
2872 e.array(1)?;
2873 }
2874 e.u64(1)?;
2875 Ok(())
2876 });
2877 assert_eq!(
2878 DataHeader::decode(&bytes).unwrap_err(),
2879 HeaderError::DepthExceeded
2880 );
2881 }
2882
2883 #[test]
2884 fn skipping_a_wide_shallow_structure_is_fine() {
2885 let bytes = encode_with(|e| {
2886 e.map(2)?;
2887 e.u64(data_key::CONTENT_LEN)?.u64(1)?;
2888 e.u64(40)?.array(64)?;
2889 for i in 0..64u64 {
2890 e.u64(i)?;
2891 }
2892 Ok(())
2893 });
2894 assert_eq!(DataHeader::decode(&bytes).unwrap().content_len, Some(1));
2895 }
2896
2897 #[test]
2900 fn duplicate_keys_are_rejected() {
2901 let bytes = encode_with(|e| {
2902 e.map(2)?;
2903 e.u64(1)?.u64(1)?;
2904 e.u64(1)?.u64(2)?;
2905 Ok(())
2906 });
2907 assert_eq!(
2908 DataHeader::decode(&bytes).unwrap_err(),
2909 HeaderError::DuplicateKey(1)
2910 );
2911 }
2912
2913 #[test]
2914 fn non_uint_keys_are_rejected() {
2915 let bytes = encode_with(|e| {
2916 e.map(1)?;
2917 e.str("endpoint")?.str("/t")?;
2918 Ok(())
2919 });
2920 assert_eq!(
2921 DataHeader::decode(&bytes).unwrap_err(),
2922 HeaderError::NonUintKey
2923 );
2924
2925 let negative = encode_with(|e| {
2926 e.map(1)?;
2927 e.i64(-1)?.u64(1)?;
2928 Ok(())
2929 });
2930 assert_eq!(
2931 DataHeader::decode(&negative).unwrap_err(),
2932 HeaderError::NonUintKey
2933 );
2934 }
2935
2936 #[test]
2937 fn indefinite_maps_are_rejected() {
2938 let bytes = encode_with(|e| {
2939 e.begin_map()?;
2940 e.u64(1)?.u64(1)?;
2941 e.end()?;
2942 Ok(())
2943 });
2944 assert_eq!(
2945 DataHeader::decode(&bytes).unwrap_err(),
2946 HeaderError::Indefinite
2947 );
2948 }
2949
2950 #[test]
2951 fn indefinite_arrays_are_rejected() {
2952 let bytes = encode_with(|e| {
2953 e.map(5)?;
2954 e.u64(0)?.begin_array()?.u64(0)?.end()?;
2955 e.u64(1)?.u64(1)?;
2956 e.u64(2)?.u64(1)?;
2957 e.u64(3)?.array(0)?;
2958 e.u64(4)?.array(0)?;
2959 Ok(())
2960 });
2961 assert_eq!(Hello::decode(&bytes).unwrap_err(), HeaderError::Indefinite);
2962 }
2963
2964 #[test]
2965 fn value_type_mismatches_are_rejected() {
2966 let bytes = encode_with(|e| {
2967 e.map(1)?;
2968 e.u64(data_key::CONTENT_LEN)?.str("not a number")?;
2969 Ok(())
2970 });
2971 assert!(matches!(
2972 DataHeader::decode(&bytes).unwrap_err(),
2973 HeaderError::Malformed(_)
2974 ));
2975 }
2976
2977 #[test]
2978 fn missing_required_keys_are_rejected() {
2979 let bytes = encode_with(|e| {
2981 e.map(1)?;
2982 e.u64(error_key::MESSAGE)?.str("why")?;
2983 Ok(())
2984 });
2985 assert_eq!(
2986 ErrorHeader::decode(&bytes).unwrap_err(),
2987 HeaderError::MissingKey(error_key::CODE)
2988 );
2989
2990 let bytes = encode_with(|e| {
2992 e.map(4)?;
2993 e.u64(0)?.array(1)?.u64(0)?;
2994 e.u64(1)?.u64(16384)?;
2995 e.u64(2)?.u64(16)?;
2996 e.u64(4)?.array(0)?;
2997 Ok(())
2998 });
2999 assert_eq!(
3000 Hello::decode(&bytes).unwrap_err(),
3001 HeaderError::MissingKey(hello_key::CAPABILITIES)
3002 );
3003 }
3004
3005 #[test]
3008 fn an_empty_filter_is_legal_and_survives_the_roundtrip() {
3009 let h = SubscriptionHeader::new("/md", "");
3010 let bytes = h.encode();
3011 assert_eq!(SubscriptionHeader::decode(&bytes).unwrap(), h);
3012 assert!(
3015 bytes.contains(&0x60),
3016 "the empty filter is encoded: {bytes:?}"
3017 );
3018 }
3019
3020 #[test]
3021 fn the_filter_grammar_accepts_what_docs_protocol_6_4_permits() {
3022 for ok in [
3023 "",
3024 "#",
3025 "px",
3026 "px.eur",
3027 "px.*",
3028 "*.eur",
3029 "sensors.*.temp",
3030 "px.#",
3031 "px.",
3032 "a..b",
3033 ] {
3034 assert_eq!(filter::validate(ok), Ok(()), "{ok:?} must be legal");
3035 }
3036 }
3037
3038 #[test]
3039 fn the_filter_grammar_rejects_partial_and_misplaced_wildcards() {
3040 for bad in [
3041 "px*", "*px", "p*x.eur", "px.e*ur", "#.px", "px.#.eur", "px#",
3042 ] {
3043 assert!(
3044 matches!(filter::validate(bad), Err(HeaderError::InvalidFilter(_))),
3045 "{bad:?} must be rejected"
3046 );
3047 }
3048 }
3049
3050 #[test]
3051 fn an_illegal_filter_is_rejected_at_the_codec_boundary() {
3052 let bytes = SubscriptionHeader::new("/md", "px.#.eur").encode();
3056 assert!(matches!(
3057 SubscriptionHeader::decode(&bytes),
3058 Err(HeaderError::InvalidFilter(_))
3059 ));
3060 let e: Error = HeaderError::InvalidFilter("`#` must be the final segment").into();
3061 assert!(e.to_string().contains("invalid topic filter"));
3062 }
3063
3064 #[test]
3065 fn subscription_strings_are_capped() {
3066 for (key, max) in [
3067 (subscription_key::ENDPOINT, limits::MAX_ENDPOINT_BYTES),
3068 (subscription_key::FILTER, limits::MAX_FILTER_BYTES),
3069 ] {
3070 let build = |len: usize| {
3071 let text = "a".repeat(len);
3072 let mut h = SubscriptionHeader::new("/md", "px.");
3073 if key == subscription_key::ENDPOINT {
3074 h.endpoint = text;
3075 } else {
3076 h.filter = text;
3077 }
3078 h.encode()
3079 };
3080 assert_eq!(
3081 SubscriptionHeader::decode(&build(max + 1)).unwrap_err(),
3082 HeaderError::StringTooLong {
3083 key,
3084 len: max + 1,
3085 max
3086 },
3087 "key {key}"
3088 );
3089 assert!(
3090 SubscriptionHeader::decode(&build(max)).is_ok(),
3091 "key {key} at cap"
3092 );
3093 }
3094 }
3095
3096 #[test]
3097 fn subscription_headers_require_both_keys() {
3098 let only = |key: u64| {
3099 encode_with(|e| {
3100 e.map(1)?;
3101 e.u64(key)?.str("/md")?;
3102 Ok(())
3103 })
3104 };
3105 assert_eq!(
3106 SubscriptionHeader::decode(&only(subscription_key::ENDPOINT)).unwrap_err(),
3107 HeaderError::MissingKey(subscription_key::FILTER)
3108 );
3109 assert_eq!(
3110 SubscriptionHeader::decode(&only(subscription_key::FILTER)).unwrap_err(),
3111 HeaderError::MissingKey(subscription_key::ENDPOINT)
3112 );
3113 }
3114
3115 #[test]
3116 fn subscription_headers_reject_malformed_input() {
3117 assert!(SubscriptionHeader::decode(&[]).is_err());
3118 let mut bytes = SubscriptionHeader::new("/md", "px.").encode();
3120 bytes.push(0xff);
3121 assert_eq!(
3122 SubscriptionHeader::decode(&bytes).unwrap_err(),
3123 HeaderError::TrailingBytes
3124 );
3125 let extended = encode_with(|e| {
3127 e.map(3)?;
3128 e.u64(0)?.str("/md")?;
3129 e.u64(1)?.str("px.")?;
3130 e.u64(40)?.array(2)?.u64(1)?.u64(2)?;
3131 Ok(())
3132 });
3133 assert_eq!(
3134 SubscriptionHeader::decode(&extended).unwrap(),
3135 SubscriptionHeader::new("/md", "px.")
3136 );
3137 }
3138
3139 #[test]
3140 fn oversized_strings_are_rejected_per_field() {
3141 let with_text = |key: u64, text: String| -> Vec<u8> {
3143 let mut h = DataHeader::reply();
3144 match key {
3145 data_key::ENDPOINT => h.endpoint = Some(text),
3146 data_key::CONTENT_TYPE => h.content_type = Some(text),
3147 data_key::TRACEPARENT => h.traceparent = Some(text),
3148 data_key::TRACESTATE => h.tracestate = Some(text),
3149 data_key::TOPIC => h.topic = Some(text),
3150 other => panic!("key {other} is not a text field"),
3151 }
3152 h.encode()
3153 };
3154 let cases: [(u64, usize); 5] = [
3155 (data_key::ENDPOINT, limits::MAX_ENDPOINT_BYTES),
3156 (data_key::CONTENT_TYPE, limits::MAX_CONTENT_TYPE_BYTES),
3157 (data_key::TRACEPARENT, limits::MAX_TRACEPARENT_BYTES),
3158 (data_key::TRACESTATE, limits::MAX_TRACESTATE_BYTES),
3159 (data_key::TOPIC, limits::MAX_TOPIC_BYTES),
3160 ];
3161 for (key, max) in cases {
3162 assert_eq!(
3163 DataHeader::decode(&with_text(key, "a".repeat(max + 1))).unwrap_err(),
3164 HeaderError::StringTooLong {
3165 key,
3166 len: max + 1,
3167 max
3168 },
3169 "key {key}"
3170 );
3171 assert!(
3172 DataHeader::decode(&with_text(key, "a".repeat(max))).is_ok(),
3173 "key {key} at cap"
3174 );
3175 }
3176 }
3177
3178 #[test]
3179 fn check_refuses_what_the_decoder_refuses() {
3180 let filter_cap = limits::MAX_FILTER_BYTES;
3181 assert!(filter::validate(&"a".repeat(filter_cap)).is_ok());
3182 assert_eq!(
3183 filter::validate(&"a".repeat(filter_cap + 1)),
3184 Err(HeaderError::StringTooLong {
3185 key: subscription_key::FILTER,
3186 len: filter_cap + 1,
3187 max: filter_cap
3188 })
3189 );
3190
3191 let cases: [(u64, usize); 5] = [
3192 (data_key::ENDPOINT, limits::MAX_ENDPOINT_BYTES),
3193 (data_key::CONTENT_TYPE, limits::MAX_CONTENT_TYPE_BYTES),
3194 (data_key::TRACEPARENT, limits::MAX_TRACEPARENT_BYTES),
3195 (data_key::TRACESTATE, limits::MAX_TRACESTATE_BYTES),
3196 (data_key::TOPIC, limits::MAX_TOPIC_BYTES),
3197 ];
3198 for (key, max) in cases {
3199 for len in [max, max + 1] {
3200 let text = "a".repeat(len);
3201 let mut data = DataHeader::reply();
3202 let mut flow = FlowHeader::new("/f", 1);
3203 match key {
3204 data_key::ENDPOINT => {
3205 data.endpoint = Some(text.clone());
3206 flow.endpoint = text;
3207 }
3208 data_key::CONTENT_TYPE => {
3209 data.content_type = Some(text.clone());
3210 flow.content_type = Some(text);
3211 }
3212 data_key::TRACEPARENT => {
3213 data.traceparent = Some(text.clone());
3214 flow.traceparent = Some(text);
3215 }
3216 data_key::TRACESTATE => {
3217 data.tracestate = Some(text.clone());
3218 flow.tracestate = Some(text);
3219 }
3220 _ => {
3221 data.topic = Some(text.clone());
3222 flow.topic = Some(text);
3223 }
3224 }
3225 assert_eq!(
3226 data.check(),
3227 DataHeader::decode(&data.encode()).map(|_| ()),
3228 "data key {key} len {len}"
3229 );
3230 assert_eq!(
3231 flow.check(),
3232 FlowHeader::decode(&flow.encode()).map(|_| ()),
3233 "flow key {key} len {len}"
3234 );
3235 }
3236 }
3237
3238 for len in [limits::MAX_ENDPOINT_BYTES, limits::MAX_ENDPOINT_BYTES + 1] {
3239 let subscription = SubscriptionHeader::new("a".repeat(len), "x");
3240 assert_eq!(
3241 subscription.check(),
3242 SubscriptionHeader::decode(&subscription.encode()).map(|_| ())
3243 );
3244 let credit = CreditHeader::new("a".repeat(len), "x", 1);
3245 assert_eq!(
3246 credit.check(),
3247 CreditHeader::decode(&credit.encode()).map(|_| ())
3248 );
3249 }
3250 }
3251
3252 #[test]
3253 fn data_check_covers_the_report_and_layer_rules() {
3254 let mut header = DataHeader::addressed("/t");
3255 header.report = vec![CursorLevel::Known(Acknowledgement::Accepted)];
3256 assert_eq!(
3257 header.check(),
3258 Err(HeaderError::InvalidReport("report without report_id"))
3259 );
3260 header.report_id = Some(1);
3261 assert_eq!(header.check(), Ok(()));
3262 header.report.clear();
3263 assert_eq!(
3264 header.check(),
3265 Err(HeaderError::InvalidReport("report_id without report"))
3266 );
3267 let mut header = DataHeader::addressed("/t");
3268 header.layer = Some(1);
3269 assert_eq!(
3270 header.check(),
3271 Err(HeaderError::InvalidLayer("layer without segment"))
3272 );
3273 header.segment = Some(0);
3274 assert_eq!(header.check(), Ok(()));
3275 header.layer = Some(limits::MAX_LAYER + 1);
3276 assert!(matches!(header.check(), Err(HeaderError::InvalidLayer(_))));
3277 }
3278
3279 #[test]
3280 fn unordered_keys_are_rejected() {
3281 let bytes = encode_with(|e| {
3284 e.map(3)?;
3285 e.u64(2)?.str("t")?;
3286 e.u64(1)?.u64(1)?;
3287 e.u64(3)?.str("p")?;
3288 Ok(())
3289 });
3290 assert_eq!(
3291 DataHeader::decode(&bytes).unwrap_err(),
3292 HeaderError::UnorderedKey(1)
3293 );
3294 }
3295
3296 #[test]
3297 fn duplicate_extension_keys_are_rejected() {
3298 let bytes = encode_with(|e| {
3299 e.map(3)?;
3300 e.u64(1)?.u64(1)?;
3301 e.u64(1000)?.u64(1)?;
3302 e.u64(1000)?.u64(2)?;
3303 Ok(())
3304 });
3305 assert_eq!(
3306 DataHeader::decode(&bytes).unwrap_err(),
3307 HeaderError::DuplicateKey(1000)
3308 );
3309 }
3310
3311 #[test]
3312 fn oversized_error_messages_are_rejected() {
3313 let big = "m".repeat(limits::MAX_MESSAGE_BYTES + 1);
3314 let bytes = encode_with(|e| {
3315 e.map(2)?;
3316 e.u64(error_key::CODE)?.u64(2)?;
3317 e.u64(error_key::MESSAGE)?.str(&big)?;
3318 Ok(())
3319 });
3320 assert_eq!(
3321 ErrorHeader::decode(&bytes).unwrap_err(),
3322 HeaderError::StringTooLong {
3323 key: error_key::MESSAGE,
3324 len: limits::MAX_MESSAGE_BYTES + 1,
3325 max: limits::MAX_MESSAGE_BYTES
3326 }
3327 );
3328 }
3329
3330 #[test]
3331 fn oversized_lists_are_rejected_without_allocating() {
3332 let bytes = encode_with(|e| {
3334 e.map(1)?;
3335 e.u64(0)?.array(u64::from(u32::MAX))?;
3336 Ok(())
3337 });
3338 assert_eq!(
3339 Hello::decode(&bytes).unwrap_err(),
3340 HeaderError::ListTooLong {
3341 key: hello_key::VERSIONS,
3342 len: u64::from(u32::MAX),
3343 max: limits::MAX_LIST_ITEMS
3344 }
3345 );
3346 }
3347
3348 #[test]
3349 fn lists_exactly_at_the_cap_are_accepted() {
3350 let bytes = encode_with(|e| {
3351 e.map(5)?;
3352 e.u64(0)?.array(limits::MAX_LIST_ITEMS as u64)?;
3353 for i in 0..limits::MAX_LIST_ITEMS as u64 {
3354 e.u64(i)?;
3355 }
3356 e.u64(1)?.u64(16384)?;
3357 e.u64(2)?.u64(16)?;
3358 e.u64(3)?.array(0)?;
3359 e.u64(4)?.array(0)?;
3360 Ok(())
3361 });
3362 assert_eq!(
3363 Hello::decode(&bytes).unwrap().versions.len(),
3364 limits::MAX_LIST_ITEMS
3365 );
3366 }
3367
3368 #[test]
3369 fn trailing_bytes_are_rejected() {
3370 let mut bytes = ErrorHeader::new(ErrorCode::Rejected).encode();
3371 bytes.push(0xff);
3372 assert_eq!(
3373 ErrorHeader::decode(&bytes).unwrap_err(),
3374 HeaderError::TrailingBytes
3375 );
3376 }
3377
3378 #[test]
3379 fn truncated_headers_are_rejected() {
3380 let full = DataHeader::addressed("/t").encode();
3381 for cut in 0..full.len() {
3382 assert!(
3383 DataHeader::decode(&full[..cut]).is_err(),
3384 "prefix of {cut} bytes must not decode"
3385 );
3386 }
3387 }
3388
3389 #[test]
3390 fn empty_input_is_rejected_for_every_header() {
3391 assert!(Hello::decode(&[]).is_err());
3392 assert!(DataHeader::decode(&[]).is_err());
3393 assert!(ErrorHeader::decode(&[]).is_err());
3394 assert!(SubscriptionHeader::decode(&[]).is_err());
3395 }
3396
3397 #[test]
3398 fn tags_are_rejected() {
3399 let bytes = encode_with(|e| {
3402 e.map(1)?;
3403 e.u64(50)?.tag(minicbor::data::IanaTag::Cbor)?.u64(1)?;
3404 Ok(())
3405 });
3406 assert_eq!(
3407 DataHeader::decode(&bytes).unwrap_err(),
3408 HeaderError::Malformed("tags are not allowed")
3409 );
3410 }
3411
3412 #[test]
3415 fn unknown_error_codes_survive_decoding() {
3416 let err = ErrorHeader {
3417 code: 99,
3418 message: None,
3419 };
3420 let decoded = ErrorHeader::decode(&err.encode()).unwrap();
3421 assert_eq!(decoded.code, 99);
3422 assert_eq!(decoded.error_code(), None);
3423 }
3424
3425 #[test]
3426 fn header_errors_become_protocol_errors() {
3427 let e: Error = HeaderError::DuplicateKey(3).into();
3428 assert!(matches!(e, Error::Protocol(_)));
3429 assert!(e.to_string().contains("duplicate header key 3"));
3430 }
3431}