Skip to main content

weida_protocol/
header.rs

1//! CBOR frame headers.
2//!
3//! The codecs are written by hand rather than derived. Three reasons:
4//!
5//! * the strictness rules (duplicate-key rejection, non-uint key rejection,
6//!   per-field string caps, depth-limited skipping) are protocol requirements,
7//!   not serialization defaults;
8//! * the exact byte layout is normative — see the golden vectors in
9//!   `docs/PROTOCOL.md` §8 — and a derive macro's field ordering is not a
10//!   contract we control;
11//! * every decoder here is a hostile-input boundary, so its allocation
12//!   behaviour has to be readable.
13//!
14//! All decode failures are protocol violations that close the connection.
15//!
16//! **No field of a DATA header is required by the decoder.** A header does not
17//! carry the stream's role, so the decoder cannot know which fields the
18//! context demands; `endpoint`-on-initiating-streams is enforced by the
19//! transport's dispatch, which does know (`docs/PROTOCOL.md` §6.2).
20
21use 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
29/// Decoder limits. The string caps are normative
30/// (`docs/PROTOCOL.md` §6); the list and depth caps are defensive
31/// implementation limits documented in the same section.
32pub mod limits {
33    /// Cap for the DATA `endpoint` field: the endpoint path cap, defined once
34    /// in `weida-core` because the local constructors enforce the same length.
35    pub const MAX_ENDPOINT_BYTES: usize = weida_core::MAX_PATH_BYTES;
36    /// Cap for the DATA `content_type` field.
37    pub const MAX_CONTENT_TYPE_BYTES: usize = 256;
38    /// Cap for the DATA `traceparent` field, defined once beside the
39    /// `traceparent` codec in `weida-core`.
40    pub const MAX_TRACEPARENT_BYTES: usize = weida_core::trace::MAX_TRACEPARENT_BYTES;
41    /// Cap for the DATA `tracestate` field, defined once beside the
42    /// `traceparent` codec in `weida-core`.
43    pub const MAX_TRACESTATE_BYTES: usize = weida_core::trace::MAX_TRACESTATE_BYTES;
44    /// Cap for the DATA `topic` field.
45    pub const MAX_TOPIC_BYTES: usize = 256;
46    /// Length of the DATA `producer` field: a raw 32-byte digest.
47    ///
48    /// Both a cap and an exact length. `docs/PROTOCOL.md` §6.2 defines the
49    /// value as "the raw 32-byte digest", so a longer one is a framing
50    /// violation and a shorter one names nothing this specification defines.
51    pub const PRODUCER_BYTES: usize = 32;
52    /// Cap for the SUBSCRIBE/UNSUBSCRIBE `filter` field.
53    pub const MAX_FILTER_BYTES: usize = 256;
54    /// Cap for the ERROR `message` field.
55    pub const MAX_MESSAGE_BYTES: usize = 1024;
56    /// Cap on the number of items in a HELLO list field.
57    ///
58    /// Without it, a hostile peer could pin `max_concurrent_uni_streams`
59    /// worth of large `Vec<u64>`s by opening many HELLO streams.
60    pub const MAX_LIST_ITEMS: usize = 64;
61    /// Cap on the number of levels a DATA header may order a report for
62    /// (`docs/PROTOCOL.md` §6.2, key `10`).
63    ///
64    /// A report order is a remote-controlled list, so it needs a cap for the
65    /// same reason [`MAX_LIST_ITEMS`] exists; 16 is more levels than the level
66    /// space defines below the application floor, so it constrains nothing a
67    /// sender legitimately wants.
68    pub const MAX_REPORT_LEVELS: usize = 16;
69    /// Nesting depth allowed when skipping an unknown field.
70    pub const MAX_SKIP_DEPTH: usize = 8;
71    /// Highest segment layer (`docs/PROTOCOL.md` §6.2 key `14`, §6.4 key
72    /// `3`).
73    pub const MAX_LAYER: u8 = 15;
74    /// Largest REPORT record map (`docs/PROTOCOL.md` §6.10).
75    pub const MAX_PATH_RECORD_BYTES: usize = 256;
76}
77
78/// HELLO keys.
79mod 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
89/// DATA keys.
90mod 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    // Key 12 (`delivery_attempt`) is reserved in docs/PROTOCOL.md §6.2;
104    // nothing codes it yet.
105    pub const SEGMENT: u64 = 13;
106    pub const LAYER: u64 = 14;
107}
108
109/// ERROR keys.
110mod error_key {
111    pub const CODE: u64 = 0;
112    pub const MESSAGE: u64 = 1;
113}
114
115/// SUBSCRIBE and UNSUBSCRIBE keys.
116mod 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
123/// CREDIT keys.
124mod credit_key {
125    pub const ENDPOINT: u64 = 0;
126    pub const FILTER: u64 = 1;
127    pub const LIMIT: u64 = 2;
128}
129
130/// CURSOR head-frame keys.
131mod cursor_key {
132    pub const REPORT_ID: u64 = 0;
133}
134
135/// FLOW keys (`docs/PROTOCOL.md` §6.8).
136mod 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
145/// The topic filter grammar of `docs/PROTOCOL.md` §6.4.
146///
147/// A topic and a filter are byte strings split on [`filter::SEPARATOR`] into
148/// segments. The two wildcards are whole-segment tokens, and everything else
149/// is literal: there is no escape character, no normalization and no case
150/// folding ([decisions/0007](../../../docs/decisions/0007-topic-namespace.md)
151/// §4.2). Both halves of the grammar live here: [`filter::validate`], the rule
152/// that says which filters may exist at all — so an illegal one is refused at
153/// the codec boundary rather than reaching a matcher that would have to cope
154/// with it — and [`filter::matches`], the matcher itself. An L2 queue selects a
155/// consumer with the same grammar as the pub/sub fan-out, and a second
156/// implementation of a wildcard language is exactly the kind of drift one
157/// definition exists to prevent.
158pub mod filter {
159    use super::HeaderError;
160
161    /// Segment separator: `.`, one byte.
162    pub const SEPARATOR: char = '.';
163    /// Matches exactly one whole segment.
164    pub const ONE_SEGMENT: &str = "*";
165    /// Matches zero or more trailing segments; legal only as the last segment.
166    pub const REST: &str = "#";
167
168    /// Does `topic` match `filter`?
169    ///
170    /// The segmented grammar of `docs/PROTOCOL.md` §6.4: segments split on `.`,
171    /// `*` for exactly one whole segment, a trailing `#` for zero or more, every
172    /// other byte literal, and the empty filter matching everything.
173    ///
174    /// Treating `*` as a wildcard makes topics with a literal `*`
175    /// unaddressable and puts a matching language in the hot path. Both are
176    /// accepted deliberately
177    /// ([decisions/0007](../../../../docs/decisions/0007-topic-namespace.md) §4.6): a
178    /// filter cannot select a segment containing `.`, `*` or `#`
179    /// literally — there is no escape character, and no sheet reports a use for
180    /// one — while a byte prefix could not express a boundary at all, so
181    /// `sensors.temp` would also select `sensors.temperature`. The hot-path half is
182    /// answered by the shape rather than by the choice: `#` is legal only as the
183    /// final segment, so this is one left-to-right walk with no backtracking, no
184    /// allocation and work bounded by the 256 B filter cap.
185    ///
186    /// A `topic` is never a pattern: `*` and `#` in a published topic are literal
187    /// bytes here, exactly like any other.
188    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                // The filter is spent: it matches only if the topic is too.
197                return topic_segments.next().is_none();
198            };
199            // Only ever the final segment — `filter::validate` rejects anything
200            // else at the codec boundary — so everything left over matches.
201            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    /// Checks `filter` against the cap and the grammar.
214    ///
215    /// The empty filter is legal and matches every topic. One pass, no
216    /// allocation. The 256 B cap is the one the decoder enforces, checked
217    /// here too so a sender refuses a filter every conformant peer would
218    /// close the connection over.
219    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
250/// Guarantee set keys (`docs/PROTOCOL.md` §6.5).
251mod 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
264/// Declares an enum whose wire form is a small `uint`, with the `core` level
265/// first so that `Default` and "absent means core" agree by construction.
266///
267/// The derived `Ord` ranks by declaration order while `to_wire` reads
268/// explicit literals, and the ladder comparisons rest on the two agreeing —
269/// `GuaranteeSet::intersect` picks the weaker level with `.min()`,
270/// `GuaranteeSet::reaches` compares with `<`, and `CursorLevel`'s derived
271/// order inherits the same coincidence — so the macro asserts the agreement
272/// at compile time rather than letting a variant inserted mid-block with a
273/// higher literal silently rank a stronger guarantee below a weaker one.
274macro_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            /// The wire value of `docs/PROTOCOL.md` §6.5.
284            pub fn to_wire(self) -> u64 {
285                match self {
286                    $($name::$variant => $value,)+
287                }
288            }
289
290            /// The level a wire value names, or `None` if the value is not
291            /// one this version defines.
292            pub fn from_wire(value: u64) -> Option<$name> {
293                match value {
294                    $($value => Some($name::$variant),)+
295                    _ => None,
296                }
297            }
298        }
299
300        // The invariant the ladder comparisons depend on. A build failure is
301        // the only acceptable outcome: at run time the mis-ranking is
302        // invisible — every value still encodes and decodes — and shows up
303        // only as a negotiated guarantee weaker than the one reported.
304        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 dimension ([`GUARANTEES.md`] §3). A ladder: later is stronger.
324    ///
325    /// [`GUARANTEES.md`]: https://github.com/tuco86/weida/blob/main/docs/GUARANTEES.md
326    Delivery {
327        /// v0: no retries, losses reported.
328        #[default]
329        BestEffort = 0,
330        /// Reserved.
331        AtMostOnce = 1,
332        /// Reserved.
333        AtLeastOnce = 2,
334    }
335}
336
337wire_enum! {
338    /// Acknowledgement/completion dimension. A ladder; the durability axes of
339    /// [`Durability`] and `replicas` are *not* part of it.
340    Acknowledgement {
341        /// Nothing is reported.
342        None = 0,
343        /// v0: QUIC's fin-acknowledgement.
344        #[default]
345        TransportReceipt = 1,
346        /// Reported by the L2 broker on the producer hop.
347        Accepted = 2,
348        /// Needs a durable store.
349        Stored = 3,
350        /// Unused.
351        Replicated = 4,
352        /// The settlement level the L2 broker orders on a delivery.
353        Processed = 5,
354    }
355}
356
357wire_enum! {
358    /// Persistence axis of `Stored`/`Replicated`
359    /// ([decisions/0004](../../../docs/decisions/0004-durability-levels.md) §4.1).
360    Durability {
361        /// Survives the broker process.
362        #[default]
363        Written = 0,
364        /// Survives loss of power on that node.
365        Flushed = 1,
366    }
367}
368
369wire_enum! {
370    /// Ordering dimension. A ladder: later is stronger.
371    OrderingMode {
372        /// v0.
373        #[default]
374        None = 0,
375        /// Report gaps, deliver as messages arrive.
376        PerProducerDetect = 1,
377        /// Hold messages back up to a bounded buffer.
378        PerProducerReassemble = 2,
379        /// L2 only.
380        PerKey = 3,
381        /// Reserved.
382        Total = 4,
383    }
384}
385
386wire_enum! {
387    /// Deduplication dimension. A ladder: later is stronger.
388    Deduplication {
389        /// v0.
390        #[default]
391        None = 0,
392        /// Suppressed within a time window.
393        Bounded = 1,
394        /// L2 only.
395        Durable = 2,
396    }
397}
398
399wire_enum! {
400    /// Backpressure dimension. **Not ordered**: these are behaviours, not
401    /// strengths, so two peers state the same one or fail to negotiate.
402    Backpressure {
403        /// v0 for Req/Rep and Push/Pull.
404        #[default]
405        Block = 0,
406        /// Refuse past a cap.
407        Reject = 1,
408        /// v0 for fan-out.
409        Drop = 2,
410        /// Reserved.
411        Spill = 3,
412        /// Reserved.
413        Coalesce = 4,
414    }
415}
416
417wire_enum! {
418    /// How a producer is named for the sequence field of
419    /// [decisions/0001](../../../docs/decisions/0001-sequence-field.md) §7.3.
420    /// **Not ordered**: two peers state the same one or fail.
421    ProducerNaming {
422        /// The proved connection fingerprint; the counter restarts with the
423        /// connection (v0 default,
424        /// [decisions/0008](../../../docs/decisions/0008-session-identity.md) §4.3).
425        #[default]
426        Fingerprint = 0,
427        /// A name supplied above L0, carried in DATA key `7`.
428        Stable = 1,
429    }
430}
431
432wire_enum! {
433    /// How often a reporter emits a record (`docs/PROTOCOL.md` §6.2, key
434    /// `11`).
435    ///
436    /// **Not ordered**: these are two shapes of the same report, not two
437    /// strengths. A cursor is never load-bearing, so neither mode is a
438    /// guarantee and neither is negotiated
439    /// ([decisions/0023](../../../docs/decisions/0023-completion-is-a-cursor.md)
440    /// §4.5).
441    ReportMode {
442        /// Records as the level advances, coalesced at the reporter's own
443        /// granularity.
444        #[default]
445        Progress = 0,
446        /// One record per level, at the end.
447        FinalOnly = 1,
448    }
449}
450
451/// A level a cursor can name: one weida defines, or one the application does.
452///
453/// The level space is **open**
454/// ([decisions/0023](../../../docs/decisions/0023-completion-is-a-cursor.md)
455/// §4.4): values below [`CursorLevel::APPLICATION_FLOOR`] are weida's own
456/// ladder, [`Acknowledgement`], and everything at or above it is an
457/// application stage weida carries and orders but never interprets. An
458/// undefined value *below* the floor is a protocol violation rather than an
459/// application level, because the reserved range is where a later version of
460/// this specification will put its own stages.
461#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
462pub enum CursorLevel {
463    /// A level this version of the protocol defines.
464    Known(Acknowledgement),
465    /// An application stage, at or above the floor.
466    Application(u64),
467}
468
469impl CursorLevel {
470    /// First wire value an application may name.
471    pub const APPLICATION_FLOOR: u64 = 16;
472
473    /// The wire value.
474    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    /// Interprets a wire value, or `None` if it is an undefined value in the
482    /// reserved range.
483    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    /// An application stage, or `None` below the floor: the reserved range is
492    /// not an application's to name.
493    pub fn application(value: u64) -> Option<CursorLevel> {
494        (value >= CursorLevel::APPLICATION_FLOOR).then_some(CursorLevel::Application(value))
495    }
496}
497
498/// One level per guarantee dimension, as declared in HELLO keys `5` and `6`
499/// (`docs/PROTOCOL.md` §6.5).
500///
501/// [`GuaranteeSet::CORE`] is the default set and is exactly what v0 does, so
502/// an absent HELLO key, an empty map and `CORE` are the same statement
503/// ([decisions/0006](../../../docs/decisions/0006-guarantee-sets.md) §4.2).
504/// Every field is a small `Copy` value: a set costs no allocation, which is
505/// what lets it ride a header a peer controls.
506#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
507pub struct GuaranteeSet {
508    /// Delivery dimension.
509    pub delivery: Delivery,
510    /// Acknowledgement/completion dimension.
511    pub acknowledgement: Acknowledgement,
512    /// Persistence axis; legal only with `Stored` or `Replicated`.
513    pub durability: Option<Durability>,
514    /// Replica count, leader included; legal only with `Replicated`, and ≥ 2.
515    pub replicas: Option<u64>,
516    /// Ordering dimension.
517    pub ordering: OrderingMode,
518    /// Deduplication dimension.
519    pub deduplication: Deduplication,
520    /// Dedup window; required with `Bounded`, forbidden otherwise.
521    pub dedup_window_ms: Option<u64>,
522    /// Backpressure behaviour. Not ordered.
523    pub backpressure: Backpressure,
524    /// How the producer of a sequenced transfer is named. Not ordered.
525    pub producer_naming: ProducerNaming,
526    /// Whether control traffic is isolated from bulk traffic
527    /// ([decisions/0002](../../../docs/decisions/0002-control-and-bulk-separation.md)
528    /// §6.1). Ordered: `true` is strictly stronger.
529    pub control_isolated: bool,
530}
531
532impl GuaranteeSet {
533    /// The default set: what v0 offers and requires.
534    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    /// Is this the default set? A `core` declaration is never written: it is
548    /// what an absent key already means.
549    pub fn is_core(&self) -> bool {
550        *self == GuaranteeSet::CORE
551    }
552
553    /// Checks the dimension combinations §6.5 forbids.
554    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    /// Writes the set as a CBOR map, omitting every dimension left at `core`.
594    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    /// Reads a set from the nested map at the decoder's position.
652    ///
653    /// The nesting is one level deep by specification (§5), and an unknown
654    /// dimension is skipped exactly like an unknown top-level key.
655    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    /// The weaker of two offered sets, dimension by dimension
698    /// (`docs/PROTOCOL.md` §2.3 step 5).
699    ///
700    /// Ladders take the minimum. Dimensions that are **not** ordered —
701    /// backpressure, producer naming, and the two independent axes of a
702    /// durability level — have no "weaker", so the two declarations must be
703    /// equal; the name of the dimension comes back as the error so a peer can
704    /// be told which one disagreed.
705    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            // A dimension the weakened acknowledgement cannot carry is
732            // dropped rather than kept: dropping is what "weaker" means here,
733            // and keeping it would produce a set §6.5 forbids. An axis one
734            // side never declared is no promise at all, so it is dropped too:
735            // carrying the other side's value would let `reaches` accept a
736            // requirement the peer never stated (§6.5, "never a quieter
737            // success"). When both declared it, the checks above made the
738            // values equal.
739            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            // A shorter window is the weaker promise.
752            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    /// Does this set reach `required` on every dimension?
768    ///
769    /// Ladders compare by level, the durability axes compare per axis, and the
770    /// unordered dimensions must match exactly. A longer dedup window is the
771    /// stronger promise.
772    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
807/// Maps a wire value to a level, naming the dimension when it is unknown.
808fn 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
812/// Maps a wire value to a segment layer, refusing one above
813/// [`limits::MAX_LAYER`] with `reason`.
814fn 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
821/// Lets [`level`] work for every dimension enum without a macro per call.
822trait 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/// Why a header was rejected. Every variant is a protocol violation.
848#[derive(Clone, Debug, PartialEq, Eq)]
849pub enum HeaderError {
850    /// The bytes are not well-formed CBOR, or a value had the wrong type.
851    Malformed(&'static str),
852    /// An indefinite-length item was used where the protocol forbids it.
853    Indefinite,
854    /// A map key appeared twice.
855    DuplicateKey(u64),
856    /// A map key was not greater than the preceding one; keys must ascend.
857    UnorderedKey(u64),
858    /// A map key was not an unsigned integer.
859    NonUintKey,
860    /// A required key was absent.
861    MissingKey(u64),
862    /// A text field exceeded its cap.
863    StringTooLong {
864        /// Key of the offending field.
865        key: u64,
866        /// Length found.
867        len: usize,
868        /// Cap for this field.
869        max: usize,
870    },
871    /// A list field declared more items than the decoder accepts.
872    ListTooLong {
873        /// Key of the offending field.
874        key: u64,
875        /// Declared item count.
876        len: u64,
877        /// Cap for list fields.
878        max: usize,
879    },
880    /// An unknown field nested deeper than [`limits::MAX_SKIP_DEPTH`].
881    DepthExceeded,
882    /// Bytes remained after the header map.
883    TrailingBytes,
884    /// A guarantee set, or a DATA header's achieved level, named a level this
885    /// version does not define.
886    UnknownLevel {
887        /// Dimension whose value was unknown.
888        dimension: &'static str,
889        /// The value found.
890        value: u64,
891    },
892    /// A guarantee set's dimension combination is one §6.5 forbids, or a
893    /// HELLO requires more than it offers (§6.1).
894    InvalidGuarantees(&'static str),
895    /// A topic filter violated the grammar of `docs/PROTOCOL.md` §6.4.
896    InvalidFilter(&'static str),
897    /// A DATA header's report order is malformed: the levels do not ascend,
898    /// there are too many of them, or the order and its id disagree
899    /// (`docs/PROTOCOL.md` §6.2, keys `9`-`11`).
900    InvalidReport(&'static str),
901    /// A segment layer is above [`limits::MAX_LAYER`], or a DATA header
902    /// carries key `14` without key `13` (`docs/PROTOCOL.md` §6.2 key `14`,
903    /// §6.4 key `3`).
904    InvalidLayer(&'static str),
905    /// An encoded header is longer than the receiver accepts (HELLO key `1`,
906    /// `docs/PROTOCOL.md` §6.1). Returned by [`crate::Agreed::check_header_len`]
907    /// on the send side; no decoder produces it, because the preamble parser
908    /// refuses an over-long header before any of it is read.
909    HeaderTooLong {
910        /// Encoded header length.
911        len: usize,
912        /// The receiver's `max_header_bytes`.
913        max: u64,
914    },
915    /// A REPORT record is longer than [`limits::MAX_PATH_RECORD_BYTES`]
916    /// (`docs/PROTOCOL.md` §6.10).
917    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
971/// Refuses `value` when it is longer than `max` bytes, naming `key`.
972fn 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
983/// Appends an encoded header to a buffer the caller owns.
984///
985/// Every `encode_into` in this file goes through here, so a hot send path can
986/// reuse one buffer and the canonical form has exactly one implementation —
987/// the property that matters, because two encoders that can disagree about
988/// key order would be a wire divergence rather than an optimisation.
989fn encode_into_with(
990    out: &mut Vec<u8>,
991    f: impl FnOnce(&mut Encoder<Vec<u8>>) -> Result<(), minicbor::encode::Error<Infallible>>,
992) {
993    // `Encoder` owns its writer, so the buffer is handed over and taken back.
994    // `std::mem::take` keeps the caller's allocation: the `Vec` that comes
995    // back is the same one, grown at most by this header.
996    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
1001/// Skips one CBOR value iteratively, refusing to recurse and refusing to nest
1002/// deeper than `max_depth`.
1003///
1004/// `minicbor` has its own `skip`, but the protocol requires a specific,
1005/// auditable bound on hostile nesting; a value that nests deeper is rejected
1006/// rather than tolerated.
1007pub fn skip_value(d: &mut Decoder<'_>, max_depth: usize) -> Result<(), HeaderError> {
1008    // `stack` holds the outstanding item counts of enclosing containers; its
1009    // length is the current nesting depth and is bounded by `max_depth`.
1010    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            // Tags, half-floats and other simple values carry no meaning in
1088            // weida headers; extensions must use plain data items.
1089            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
1107/// Reader for one strict header map.
1108struct MapReader<'a, 'b> {
1109    d: &'a mut Decoder<'b>,
1110    remaining: u64,
1111    /// Bitmask of seen keys `0..=63`, for the required-key checks. The
1112    /// specification reserves that range, so presence needs no allocation.
1113    seen: u64,
1114    /// Previously read key.
1115    ///
1116    /// Keys are required to ascend strictly, which makes duplicate detection
1117    /// complete for *every* key — including extension keys the decoder skips —
1118    /// in constant space. A set of seen extension keys would be exactly the
1119    /// remote-controlled allocation the invariants forbid.
1120    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    /// Reads a byte string of exactly `N` bytes.
1197    ///
1198    /// The cap is checked before the length is trusted for anything, and a
1199    /// shorter value is rejected rather than padded: `docs/PROTOCOL.md` §6.2
1200    /// defines the one field that uses this as a raw 32-byte digest, and half
1201    /// a digest identifies nobody.
1202    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        // `len` is now bounded by MAX_LIST_ITEMS, so reserving is safe.
1233        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    /// Reads a report order: a definite-length array of strictly ascending
1241    /// cursor levels, capped at [`limits::MAX_REPORT_LEVELS`].
1242    ///
1243    /// Ascent is checked here rather than after the fact for the same reason
1244    /// map keys are: it makes duplicate detection complete in constant space,
1245    /// and it makes the wire form canonical, so two peers ordering the same
1246    /// levels send the same bytes.
1247    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        // `len` is now bounded by MAX_REPORT_LEVELS, so reserving is safe.
1257        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
1282/// Rejects trailing bytes after a header map.
1283fn finish(d: &Decoder<'_>) -> Result<(), HeaderError> {
1284    if d.position() == d.input().len() {
1285        Ok(())
1286    } else {
1287        Err(HeaderError::TrailingBytes)
1288    }
1289}
1290
1291/// HELLO header: connection negotiation input.
1292#[derive(Clone, Debug, PartialEq, Eq)]
1293pub struct Hello {
1294    /// Wire protocol versions the sender supports.
1295    pub versions: Vec<u64>,
1296    /// Largest header the sender is willing to receive.
1297    pub max_header_bytes: u64,
1298    /// Advisory concurrent inbound transfer count.
1299    pub max_transfers: u64,
1300    /// Capability codes the sender supports.
1301    pub capabilities: Vec<u64>,
1302    /// Capability codes the sender requires the peer to support.
1303    pub required_capabilities: Vec<u64>,
1304    /// Guarantee set the sender can honour (key `5`).
1305    ///
1306    /// `None` means the default set: an absent key and
1307    /// [`GuaranteeSet::CORE`] are the same declaration, which is why a v0
1308    /// HELLO is unchanged on the wire.
1309    pub guarantees_offered: Option<GuaranteeSet>,
1310    /// Guarantee set the sender requires of the peer (key `6`).
1311    ///
1312    /// MUST be reachable by `guarantees_offered` on every dimension: requiring
1313    /// what you cannot honour yourself is a configuration error
1314    /// (`docs/PROTOCOL.md` §6.1), and a decoder rejects it.
1315    pub guarantees_required: Option<GuaranteeSet>,
1316}
1317
1318impl Hello {
1319    /// The HELLO a v0 implementation sends: no guarantee declarations, so
1320    /// `core` offered and `core` required.
1321    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    /// The set this HELLO offers; an absent declaration means `core`.
1334    pub fn offered(&self) -> GuaranteeSet {
1335        self.guarantees_offered.unwrap_or(GuaranteeSet::CORE)
1336    }
1337
1338    /// The set this HELLO requires; an absent declaration means `core`.
1339    pub fn required(&self) -> GuaranteeSet {
1340        self.guarantees_required.unwrap_or(GuaranteeSet::CORE)
1341    }
1342
1343    /// Allocating form of [`Self::encode_into`].
1344    pub fn encode(&self) -> Vec<u8> {
1345        let mut out = Vec::new();
1346        self.encode_into(&mut out);
1347        out
1348    }
1349
1350    /// Appends the encoded header to `out`, for a send path that reuses a
1351    /// buffer.
1352    pub fn encode_into(&self, out: &mut Vec<u8>) {
1353        encode_into_with(out, |e| {
1354            // A `core` declaration is never written: an absent key already
1355            // says it, and a v0 HELLO must stay byte-identical (§6.1).
1356            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    /// Decodes the header.
1390    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        // §6.1: requiring more than you offer is a configuration error, and
1438        // one a decoder can see in a single header.
1439        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/// DATA header: one transfer.
1449///
1450/// Every field is optional at the decoder. Which of them the *context*
1451/// requires is a dispatch question: an initiating stream must name an endpoint
1452/// and the reply half of an exchange must not, but the decoder sees bytes, not
1453/// streams (`docs/PROTOCOL.md` §6.2).
1454#[derive(Clone, Debug, Default, PartialEq, Eq)]
1455pub struct DataHeader {
1456    /// Endpoint path. Required on an initiating stream, ignored on a reply.
1457    pub endpoint: Option<String>,
1458    /// Advisory payload length.
1459    pub content_len: Option<u64>,
1460    /// Opaque content type label.
1461    pub content_type: Option<String>,
1462    /// W3C `traceparent`.
1463    pub traceparent: Option<String>,
1464    /// W3C `tracestate`, opaque passthrough.
1465    pub tracestate: Option<String>,
1466    /// Pub/Sub topic; opaque bytes, selected by the filter grammar of
1467    /// `docs/PROTOCOL.md` §6.4. Only meaningful on transfers fanned out by a
1468    /// publisher.
1469    pub topic: Option<String>,
1470    /// Per-producer sequence number, for ordering and gap detection
1471    /// (`docs/PROTOCOL.md` §6.2, key `6`).
1472    ///
1473    /// Written by a publisher whose connection negotiated `PerProducer`
1474    /// ordering, and by nothing under `core`: the number is assigned once per
1475    /// published message, before fan-out, so a copy a subscriber lost shows up
1476    /// as a hole in its own sequence. It is not a transfer identifier and
1477    /// correlates nothing — an exchange is correlated by its stream.
1478    pub sequence: Option<u64>,
1479    /// Producer identity: the raw 32-byte digest (`docs/PROTOCOL.md` §6.2,
1480    /// key `7`).
1481    ///
1482    /// **Specified ahead of code**, and absent in the default case by design:
1483    /// the receiver already knows the sending peer's proved fingerprint from
1484    /// the handshake, so this names a producer only where it is *not* the
1485    /// connection peer — a relay, or a name an L2 subscription supplies
1486    /// ([decisions/0008](../../../docs/decisions/0008-session-identity.md)
1487    /// §4.4). The `sha256:<64 hex>` spelling is presentation only and never
1488    /// goes on the wire.
1489    pub producer: Option<[u8; limits::PRODUCER_BYTES]>,
1490    /// The completion level the sender **achieved** for the message it is
1491    /// answering (`docs/PROTOCOL.md` §6.2, key `8`).
1492    ///
1493    /// This is the L2 confirm, and it is a statement about one hop: a broker
1494    /// that has taken responsibility for a message in memory writes
1495    /// [`Acknowledgement::Accepted`] on the reply half of the producer's
1496    /// exchange, which is what makes the reply a publisher confirm without a
1497    /// frame kind of its own
1498    /// ([decisions/0018](../../../docs/decisions/0018-minimal-broker.md)
1499    /// §4.6). It is *achieved*, never requested — a level a peer wants is
1500    /// negotiated in HELLO and refused there if it cannot be reached
1501    /// ([0006](../../../docs/decisions/0006-guarantee-sets.md) §4.4) — and it
1502    /// is never relayed: the producer's confirm says nothing about what a
1503    /// consumer later does with the message
1504    /// ([GUARANTEES.md] §2).
1505    ///
1506    /// A v0 sender leaves it absent, and an absent key is not
1507    /// `Acknowledgement::None`: it says this hop makes no claim beyond the
1508    /// transport receipt QUIC already gave.
1509    ///
1510    /// [GUARANTEES.md]: https://github.com/tuco86/weida/blob/main/docs/GUARANTEES.md
1511    pub achieved: Option<Acknowledgement>,
1512    /// Identifier the sender assigns to the report it orders (key `9`).
1513    ///
1514    /// Present exactly when [`DataHeader::report`] is non-empty. It names the
1515    /// CURSOR stream that will report on *this* transfer, and it is scoped to
1516    /// the connection and to the direction that allocated it: a peer reports
1517    /// only on transfers it received, so the two directions cannot collide.
1518    pub report_id: Option<u64>,
1519    /// Levels the sender asks to be reported, strictly ascending (key `10`).
1520    ///
1521    /// An **order**, not a guarantee: a receiver that cannot reach a level
1522    /// simply does not report it, and the transfer does not fail for it. A
1523    /// level a peer must reach is the negotiated `acknowledgement` dimension
1524    /// of HELLO instead
1525    /// ([decisions/0006](../../../docs/decisions/0006-guarantee-sets.md)
1526    /// §4.4).
1527    pub report: Vec<CursorLevel>,
1528    /// How often the reporter should emit a record (key `11`).
1529    ///
1530    /// [`ReportMode::Progress`] is the default and is never written.
1531    pub report_mode: ReportMode,
1532    /// Segment number per `(sender, path, topic)`, from 0 (key `13`).
1533    ///
1534    /// Written by a radio or by `Peer::segment`
1535    /// ([decisions/0034](../../../docs/decisions/0034-late-is-lost.md)
1536    /// §4.6, [decisions/0037](../../../docs/decisions/0037-layered-segments.md)
1537    /// §4.2). It is not a `PerProducer` sequence: it is written whatever
1538    /// ordering was negotiated, and a dish uses it to discard a segment older
1539    /// than the newest it delivered.
1540    pub segment: Option<u64>,
1541    /// Segment layer, `0..=15` (key `14`); absent means 0; written only
1542    /// beside `segment`
1543    /// ([decisions/0037](../../../docs/decisions/0037-layered-segments.md)
1544    /// §4.3). The encoder writes what it is given, so a sender passes `None`
1545    /// for layer 0.
1546    pub layer: Option<u8>,
1547}
1548
1549impl DataHeader {
1550    /// A header addressing `endpoint`, for the initiating half of a stream.
1551    pub fn addressed(endpoint: impl Into<String>) -> DataHeader {
1552        DataHeader {
1553            endpoint: Some(endpoint.into()),
1554            ..DataHeader::default()
1555        }
1556    }
1557
1558    /// A header for the reply half of an exchange: no endpoint, no topic.
1559    ///
1560    /// The stream is the correlation, so a reply carries no identifier of the
1561    /// request it answers.
1562    pub fn reply() -> DataHeader {
1563        DataHeader::default()
1564    }
1565
1566    /// Checks the rules the decoder enforces, so a sender can refuse a header
1567    /// before it makes every conformant peer close the connection.
1568    ///
1569    /// Covers the text caps of [`limits`], the layer range and its need for a
1570    /// segment, and the report rules [`DataHeader::encode`] cannot report: at
1571    /// most [`limits::MAX_REPORT_LEVELS`] distinct levels, and key `9` present
1572    /// exactly when key `10` is. Whether the encoded header fits the peer's
1573    /// `max_header_bytes` is [`crate::Agreed::check_header_len`].
1574    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            // The encoder writes distinct levels, so the cap counts those.
1613            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    /// Encodes the header.
1626    ///
1627    /// Key `10` goes out in the canonical form §6.2 makes normative —
1628    /// strictly ascending by wire value, no repeats — whatever order
1629    /// [`DataHeader::report`] happens to hold. That rule is enforced here
1630    /// because encoding cannot fail: a vector in any other order would
1631    /// otherwise produce bytes that close the connection at every conformant
1632    /// peer, and there would be no way to tell the caller so.
1633    ///
1634    /// The report's other two rules stay the caller's for exactly that
1635    /// reason — both need an error, and this function has none to give;
1636    /// [`DataHeader::check`] is where they fail. At most
1637    /// [`limits::MAX_REPORT_LEVELS`] distinct levels, and key `9` present
1638    /// exactly when key `10` is.
1639    ///
1640    /// Allocating form of [`Self::encode_into`].
1641    pub fn encode(&self) -> Vec<u8> {
1642        let mut out = Vec::new();
1643        self.encode_into(&mut out);
1644        out
1645    }
1646
1647    /// Appends the encoded header to `out`, for a send path that reuses a
1648    /// buffer.
1649    pub fn encode_into(&self, out: &mut Vec<u8>) {
1650        // Sorted and deduplicated by wire value, not by the enum's derived
1651        // order, because the wire value is what ascends on the wire.
1652        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            // Keys 6 and 7 are written only when set, which for every v0
1690            // sender means never: nothing in `weida` populates them yet.
1691            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            // `Progress` is never written: an absent key already says it, so
1710            // a header that orders a report in the default mode stays as
1711            // short as the mode is uninteresting (§6.5's rule for levels).
1712            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    /// Decodes the header.
1727    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                    // An unknown level is not a level: a peer naming one this
1751                    // version does not define is refused rather than silently
1752                    // read as the weakest, because the value decides what a
1753                    // producer believes about its message.
1754                    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        // A layer belongs to a segment: key 14 alone names a layer of
1776        // nothing.
1777        if header.layer.is_some() && header.segment.is_none() {
1778            return Err(HeaderError::InvalidLayer("layer without segment"));
1779        }
1780        // Keys 9 and 10 are one statement in two halves: an order with no
1781        // stream to report on, or a stream with nothing to report, names a
1782        // report nobody can serve.
1783        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/// ERROR header.
1794///
1795/// Legal only on the reply half of a bidirectional stream: an ERROR is the
1796/// alternative to a reply, so it needs no reference to what it answers.
1797#[derive(Clone, Debug, PartialEq, Eq)]
1798pub struct ErrorHeader {
1799    /// Raw error code.
1800    pub code: u64,
1801    /// Human-readable detail; never machine-interpreted.
1802    pub message: Option<String>,
1803}
1804
1805impl ErrorHeader {
1806    /// Builds a header for a known error code.
1807    pub fn new(code: weida_core::ErrorCode) -> ErrorHeader {
1808        ErrorHeader {
1809            code: code.to_wire(),
1810            message: None,
1811        }
1812    }
1813
1814    /// The error code, or `None` for an unknown one.
1815    pub fn error_code(&self) -> Option<weida_core::ErrorCode> {
1816        weida_core::ErrorCode::from_wire(self.code)
1817    }
1818
1819    /// Allocating form of [`Self::encode_into`].
1820    pub fn encode(&self) -> Vec<u8> {
1821        let mut out = Vec::new();
1822        self.encode_into(&mut out);
1823        out
1824    }
1825
1826    /// Appends the encoded header to `out`, for a send path that reuses a
1827    /// buffer.
1828    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    /// Decodes the header.
1841    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/// SUBSCRIBE and UNSUBSCRIBE header.
1862///
1863/// Both frames carry the same two keys: the publisher path to (un)subscribe on
1864/// and the topic filter. The filter is a segmented pattern, not a byte prefix
1865/// ([`filter`]): the empty filter matches every topic, `*` matches one whole
1866/// segment and a trailing `#` matches zero or more.
1867#[derive(Clone, Debug, PartialEq, Eq)]
1868pub struct SubscriptionHeader {
1869    /// Publisher endpoint path.
1870    pub endpoint: String,
1871    /// Topic filter; the empty string matches everything. A decoded header's
1872    /// filter has passed [`filter::validate`].
1873    pub filter: String,
1874    /// The dish's latency budget in milliseconds (key `2`), meaningful on a
1875    /// RADIO path only
1876    /// ([decisions/0034](../../../docs/decisions/0034-late-is-lost.md)
1877    /// §4.6). Encoded only when present.
1878    pub max_age_ms: Option<u64>,
1879    /// The highest layer the dish wants (key `3`), `0..=15`, meaningful on a
1880    /// RADIO path only; absent means no cap
1881    /// ([decisions/0037](../../../docs/decisions/0037-layered-segments.md)
1882    /// §4.3). Encoded only when present.
1883    pub max_layer: Option<u8>,
1884}
1885
1886impl SubscriptionHeader {
1887    /// A header for `endpoint` and `filter`.
1888    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    /// Checks the endpoint and filter caps, the filter grammar and the layer
1898    /// range: everything the decoder refuses that the encoder cannot.
1899    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    /// Allocating form of [`Self::encode_into`].
1916    pub fn encode(&self) -> Vec<u8> {
1917        let mut out = Vec::new();
1918        self.encode_into(&mut out);
1919        out
1920    }
1921
1922    /// Appends the encoded header to `out`, for a send path that reuses a
1923    /// buffer.
1924    pub fn encode_into(&self, out: &mut Vec<u8>) {
1925        // Keys 0 and 1 are required, so neither is elided: an absent filter
1926        // and an empty filter would otherwise be indistinguishable on the
1927        // wire, and the empty filter is the "everything" subscription. Keys 2
1928        // and 3 are optional and written only when set.
1929        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    /// Decodes the header.
1945    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        // The grammar is checked here, at the codec boundary, so no matcher
1972        // ever sees a filter it would have to interpret twice; an illegal one
1973        // closes the connection with `PROTOCOL_VIOLATION`
1974        // (`docs/PROTOCOL.md` §6.4).
1975        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/// FLOW header (kind `7`).
1988///
1989/// The registration of a datagram flow: the stream it opens is the flow's
1990/// lifetime, and the flow id prefixes every datagram the flow carries
1991/// ([decisions/0034](../../../docs/decisions/0034-late-is-lost.md) §4.2).
1992/// The optional keys reuse DATA's meaning and caps.
1993#[derive(Clone, Debug, PartialEq, Eq)]
1994pub struct FlowHeader {
1995    /// Endpoint path the flow is addressed to (key `0`, required).
1996    pub endpoint: String,
1997    /// Flow id, chosen by the sender and unique per connection and
1998    /// direction (key `1`, required).
1999    pub flow: u64,
2000    /// Opaque media type label (key `2`).
2001    pub content_type: Option<String>,
2002    /// W3C Trace Context `traceparent` (key `3`).
2003    pub traceparent: Option<String>,
2004    /// W3C Trace Context `tracestate` (key `4`).
2005    pub tracestate: Option<String>,
2006    /// Topic the flow carries (key `5`).
2007    pub topic: Option<String>,
2008}
2009
2010impl FlowHeader {
2011    /// A header for flow `flow` addressed to `endpoint`, with no optional key.
2012    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    /// Checks the text caps of the decoder, so a sender can refuse a header
2024    /// every conformant peer would close the connection over.
2025    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    /// Allocating form of [`Self::encode_into`].
2054    pub fn encode(&self) -> Vec<u8> {
2055        let mut out = Vec::new();
2056        self.encode_into(&mut out);
2057        out
2058    }
2059
2060    /// Appends the encoded header to `out`, for a send path that reuses a
2061    /// buffer.
2062    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    /// Decodes the header.
2089    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/// CREDIT header (kind `5`).
2132///
2133/// The L2 credit of
2134/// [decisions/0003](../../../docs/decisions/0003-credit-unit.md) §4.2-§4.3:
2135/// which subscription, and how many messages that subscription will accept in
2136/// total. Three keys, all required — a subscription is `(endpoint, filter)`
2137/// on the connection the frame arrives on, and an absent limit would be
2138/// indistinguishable from a limit of zero, which is the pause.
2139///
2140/// **The limit is absolute and cumulative, not a delta.** It counts messages
2141/// delivered on that subscription since it was created, so a lost frame costs
2142/// nothing and a duplicated one changes nothing. It is also **monotone at the
2143/// receiver**: a broker keeps the highest limit it has seen, which is what
2144/// makes a reordered frame harmless on a transport that does not order the
2145/// streams control frames ride. Monotone is the whole rule: a receiver
2146/// ignores any limit that is not strictly greater than the standing one, so
2147/// restating a number already delivered changes nothing unless the
2148/// subscription had exhausted its credit anyway. **v0 offers no way to lower
2149/// a standing limit.** The only pause is the `0` a fresh subscription starts
2150/// at, so a consumer that wants to stay in control grants in increments it
2151/// is willing to receive. AMQP 1.0 can shrink `link-credit` against an
2152/// absolute baseline [amqp10 §5.1]; this frame cannot, and a peer that reads
2153/// it as if it could would wait for a stop no broker can deliver.
2154#[derive(Clone, Debug, PartialEq, Eq)]
2155pub struct CreditHeader {
2156    /// Endpoint path of the queue the subscription is on.
2157    pub endpoint: String,
2158    /// Topic filter of the subscription. A decoded header's filter has passed
2159    /// [`filter::validate`].
2160    pub filter: String,
2161    /// Messages this subscription will accept in total, counted from its
2162    /// creation.
2163    pub limit: u64,
2164}
2165
2166impl CreditHeader {
2167    /// A header granting `limit` to the subscription `(endpoint, filter)`.
2168    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    /// Checks the endpoint cap, the filter cap and the filter grammar.
2177    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    /// Allocating form of [`Self::encode_into`].
2187    pub fn encode(&self) -> Vec<u8> {
2188        let mut out = Vec::new();
2189        self.encode_into(&mut out);
2190        out
2191    }
2192
2193    /// Appends the encoded header to `out`, for a send path that reuses a
2194    /// buffer.
2195    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    /// Decodes the header.
2206    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        // Same boundary as SUBSCRIBE: a filter that does not name a
2228        // subscription cannot grant credit to one.
2229        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/// CURSOR head frame (kind `6`).
2241///
2242/// One key, required: which report this stream carries. The stream then
2243/// carries `(level, offset)` records until FIN and **never any payload**
2244/// ([decisions/0024](../../../docs/decisions/0024-three-families-one-back-channel.md)
2245/// §4.4) — which is why the head frame needs no length, no endpoint and no
2246/// correlation beyond the id the DATA header allocated.
2247#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2248pub struct CursorHeader {
2249    /// The `report_id` of the DATA header that ordered this report.
2250    pub report_id: u64,
2251}
2252
2253impl CursorHeader {
2254    /// Allocating form of [`Self::encode_into`].
2255    pub fn encode(&self) -> Vec<u8> {
2256        let mut out = Vec::new();
2257        self.encode_into(&mut out);
2258        out
2259    }
2260
2261    /// Appends the encoded header to `out`, for a send path that reuses a
2262    /// buffer.
2263    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    /// Decodes the header.
2272    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
2292/// Longest possible cursor record: two 8-byte QUIC varints.
2293pub const MAX_CURSOR_RECORD_LEN: usize = 2 * crate::varint::MAX_ENCODED_LEN;
2294
2295/// Appends one `(level, offset)` record to `out`.
2296///
2297/// Records are QUIC varint pairs rather than CBOR: a record is hot-path and
2298/// self-delimiting, and a CBOR map per record would cost a map header per
2299/// reported byte range for no gain — the head frame already carries every
2300/// field a record could need to name.
2301pub 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
2310/// Decodes one record from the front of `input`.
2311///
2312/// `Ok(None)` means the input ends inside a record: read more bytes and
2313/// retry. That is **not** a violation — a reader sees whatever slice the
2314/// transport handed it, and a record is at most
2315/// [`MAX_CURSOR_RECORD_LEN`] bytes, so the retry is bounded. An undefined
2316/// level in the reserved range *is* a violation: the value decides what the
2317/// receiver believes about its own transfer.
2318pub fn decode_cursor_record(
2319    input: &[u8],
2320) -> Result<Option<(CursorLevel, u64, usize)>, HeaderError> {
2321    // `decode_varint` accepts the whole representable range, so its only
2322    // failure is a value the input ended inside of.
2323    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/// REPORT head frame (kind `8`): the empty map, every key reserved
2337/// (`docs/PROTOCOL.md` §6.10). Records follow it until FIN.
2338#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
2339pub struct ReportHead;
2340
2341impl ReportHead {
2342    /// Encodes the head frame: `A0`.
2343    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    /// Decodes the head frame; unknown keys are skipped by §5's rule.
2353    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
2366/// REPORT record keys (`docs/PROTOCOL.md` §6.10).
2367mod 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/// One REPORT record: the sender's view of the QUIC path under the
2383/// connection (`docs/PROTOCOL.md` §6.10,
2384/// [decisions/0036](../../../docs/decisions/0036-connection-statistics.md)
2385/// §4.5). Numbers only; no address travels. A key the peer did not send
2386/// reads as `0`, and the encoder writes only the non-zero ones.
2387#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
2388pub struct PathRecord {
2389    /// Smoothed round-trip time, µs (key `0`).
2390    pub rtt_us: u64,
2391    /// Smallest round-trip time seen, µs (key `1`).
2392    pub min_rtt_us: u64,
2393    /// Congestion window, bytes (key `2`).
2394    pub cwnd: u64,
2395    /// Congestion events the sender's controller reacted to (key `3`).
2396    pub congestion_events: u64,
2397    /// Packets the sender sent and declared lost (key `4`).
2398    pub lost_packets: u64,
2399    /// Bytes the sender sent and declared lost (key `5`).
2400    pub lost_bytes: u64,
2401    /// Packets the sender sent (key `6`).
2402    pub sent_packets: u64,
2403    /// The largest UDP payload the path carries now (key `7`).
2404    pub current_mtu: u64,
2405    /// UDP datagrams the sender sent (key `8`).
2406    pub tx_datagrams: u64,
2407    /// Bytes in them (key `9`).
2408    pub tx_bytes: u64,
2409    /// UDP datagrams the sender received (key `10`).
2410    pub rx_datagrams: u64,
2411    /// Bytes in them (key `11`).
2412    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    /// Appends `[varint length][CBOR map]` to `out`. Twelve keys of at most
2434    /// nine bytes each stay far below [`limits::MAX_PATH_RECORD_BYTES`].
2435    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    /// Decodes one record from the front of `input`.
2453    ///
2454    /// `Ok(None)` means the input ends inside a record: read more bytes and
2455    /// retry, which is bounded because a record is at most
2456    /// [`limits::MAX_PATH_RECORD_BYTES`] bytes plus its prefix; at FIN the
2457    /// same answer is a truncated record and a violation. A length prefix
2458    /// above the cap is refused before any of the map is read.
2459    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    /// The tests build headers by hand; production code goes through each
2504    /// type's `encode_into`.
2505    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    // --- golden vectors, docs/PROTOCOL.md §8 ------------------------------
2519
2520    #[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        // The stream is the correlation, so a reply header is an empty map.
2532        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    /// The digest of the §8 vectors: SHA-256 of `"test"`, the value the
2579    /// address examples in `docs/PROTOCOL.md` already use.
2580    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        // Half a digest identifies nobody, so it is a framing violation
2630        // rather than a value to carry (`docs/PROTOCOL.md` §6.2).
2631        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        // What the runtime writes on a `core` connection: neither key 6 —
2662        // which needs negotiated `PerProducer` ordering — nor key 7, which
2663        // nothing in this repository sets.
2664        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        // One header layout serves both kinds; only the kind byte differs, and
2691        // that byte belongs to the preamble (see `tests/golden_vectors.rs`).
2692        assert_eq!(SubscriptionHeader::decode(&bytes).unwrap(), h);
2693    }
2694
2695    // --- roundtrips -------------------------------------------------------
2696
2697    #[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        // A caller hands over the levels in the order it thought of them.
2770        // Decoding is the proof: the decoder refuses a non-ascending or
2771        // repeated array, so a header that survives its own encoder was
2772        // canonicalized on the way out.
2773        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    // --- optional fields --------------------------------------------------
2796
2797    #[test]
2798    fn every_data_field_is_optional_at_the_decoder() {
2799        // The endpoint requirement lives in dispatch, not here: a reply half
2800        // legitimately carries none, and the decoder cannot tell the halves
2801        // apart.
2802        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    // --- forward compatibility -------------------------------------------
2821
2822    #[test]
2823    fn unknown_keys_are_skipped() {
2824        // Re-encode the golden DATA header with an extra key 63 holding a
2825        // nested structure, and check it still decodes to the same value.
2826        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    // --- strictness -------------------------------------------------------
2898
2899    #[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        // ERROR without a code.
2980        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        // HELLO missing capabilities.
2991        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    // --- subscription headers ---------------------------------------------
3006
3007    #[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        // The key is written even though the value is empty: absent and empty
3013        // must stay distinguishable, and empty means "every topic".
3014        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        // Encoding does not validate — a test may build any bytes — but
3053        // decoding does, which is what makes the grammar enforceable against
3054        // a peer (`docs/PROTOCOL.md` §6.4).
3055        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        // Trailing bytes.
3119        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        // Unknown keys are skipped, like every other header.
3126        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        // Built through the encoder, which emits keys in ascending order.
3142        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        // Descending keys break the ascending-order rule, which is what makes
3282        // duplicate detection complete for extension keys.
3283        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        // A one-byte array header claiming 2^32 items must not reserve memory.
3333        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        // Key 50 is unknown, so the value goes through `skip_value`, which is
3400        // where the tag rule lives.
3401        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    // --- reserved value passthrough ---------------------------------------
3413
3414    #[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}