weida_protocol/
frame.rs

1//! Stream preamble: `[0x57][kind][header_len varint][CBOR header]`.
2//!
3//! Every unidirectional stream in both directions starts with this preamble.
4//! Parsing it is the first thing that touches hostile bytes, so it is the place
5//! where the header size cap is enforced — before any allocation.
6
7use std::fmt;
8
9use crate::varint::{self, VarintError, decode_varint};
10
11/// First byte of every weida stream: ASCII `W`.
12pub const MAGIC: u8 = 0x57;
13
14/// Longest possible preamble: magic, kind and an 8-byte varint.
15pub const MAX_PREAMBLE_LEN: usize = 2 + varint::MAX_ENCODED_LEN;
16
17/// What a stream carries.
18#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
19pub enum FrameKind {
20    /// Connection negotiation; header-only.
21    Hello,
22    /// A transfer: header followed by opaque payload until FIN.
23    Data,
24    /// Failure report on the reply half of an exchange; header-only.
25    Error,
26    /// Registration of interest in a publisher's topics; header-only.
27    Subscribe,
28    /// Withdrawal of a previous SUBSCRIBE; header-only.
29    Unsubscribe,
30    /// An L2 consumer's delivery limit for one subscription; header-only.
31    ///
32    /// The credit of
33    /// [decisions/0003](../../../docs/decisions/0003-credit-unit.md) §4.2: a
34    /// subscription and an **absolute** delivery limit, so a lost or
35    /// duplicated frame changes nothing. Only a queue serves it; a path with
36    /// no queue refuses the stream with `UNSUPPORTED`.
37    Credit,
38    /// A report about one payload stream; header-only followed by records.
39    ///
40    /// A head frame naming the stream, then `(level, offset)` records until
41    /// FIN. Never shares a stream with payload
42    /// ([decisions/0024](../../../docs/decisions/0024-three-families-one-back-channel.md)
43    /// §4.4).
44    Cursor,
45    /// A datagram flow's registration; header-only on QUIC, held open until
46    /// FIN.
47    ///
48    /// The stream is the flow's lifetime: FIN closes it, `RESET_STREAM`
49    /// abandons it, `STOP_SENDING` refuses it. The units travel as QUIC
50    /// DATAGRAM frames prefixed with the flow id, and on a local transport as
51    /// length-prefixed records on this stream
52    /// ([decisions/0034](../../../docs/decisions/0034-late-is-lost.md) §4.2).
53    Flow,
54    /// One side's view of the QUIC path, for the peer; a head frame, then
55    /// length-prefixed records until FIN. QUIC only, and only when both
56    /// HELLOs listed capability code `2`
57    /// ([decisions/0036](../../../docs/decisions/0036-connection-statistics.md)
58    /// §4.5).
59    Report,
60}
61
62impl FrameKind {
63    /// The wire code.
64    pub const fn to_u8(self) -> u8 {
65        match self {
66            FrameKind::Hello => 0,
67            FrameKind::Data => 1,
68            FrameKind::Error => 2,
69            FrameKind::Subscribe => 3,
70            FrameKind::Unsubscribe => 4,
71            FrameKind::Credit => 5,
72            FrameKind::Cursor => 6,
73            FrameKind::Flow => 7,
74            FrameKind::Report => 8,
75        }
76    }
77
78    /// Interprets a wire code. Unknown kinds are a protocol violation, not a
79    /// forward-compatibility hook: a receiver cannot know whether an unknown
80    /// stream kind carries payload it would have to drain.
81    pub const fn from_u8(code: u8) -> Option<FrameKind> {
82        match code {
83            0 => Some(FrameKind::Hello),
84            1 => Some(FrameKind::Data),
85            2 => Some(FrameKind::Error),
86            3 => Some(FrameKind::Subscribe),
87            4 => Some(FrameKind::Unsubscribe),
88            5 => Some(FrameKind::Credit),
89            6 => Some(FrameKind::Cursor),
90            7 => Some(FrameKind::Flow),
91            8 => Some(FrameKind::Report),
92            _ => None,
93        }
94    }
95
96    /// True if payload bytes follow the header until FIN.
97    pub const fn has_payload(self) -> bool {
98        matches!(self, FrameKind::Data)
99    }
100}
101
102impl fmt::Display for FrameKind {
103    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
104        let s = match self {
105            FrameKind::Hello => "HELLO",
106            FrameKind::Data => "DATA",
107            FrameKind::Error => "ERROR",
108            FrameKind::Subscribe => "SUBSCRIBE",
109            FrameKind::Unsubscribe => "UNSUBSCRIBE",
110            FrameKind::Credit => "CREDIT",
111            FrameKind::Cursor => "CURSOR",
112            FrameKind::Flow => "FLOW",
113            FrameKind::Report => "REPORT",
114        };
115        f.write_str(s)
116    }
117}
118
119/// A parsed stream preamble.
120#[derive(Clone, Copy, Debug, PartialEq, Eq)]
121pub struct Preamble {
122    /// Stream kind.
123    pub kind: FrameKind,
124    /// Length of the CBOR header that follows, in bytes.
125    pub header_len: u64,
126}
127
128/// Why a preamble was rejected.
129#[derive(Clone, Copy, Debug, PartialEq, Eq)]
130pub enum PreambleError {
131    /// Not enough bytes yet. Read more and retry; **not** a protocol violation.
132    Incomplete,
133    /// The first byte was not [`MAGIC`].
134    BadMagic(u8),
135    /// The kind byte is not defined in this protocol version.
136    UnknownKind(u8),
137    /// The advertised header length exceeds the local cap. Reported before any
138    /// allocation is attempted.
139    HeaderTooLarge {
140        /// Length the peer advertised.
141        len: u64,
142        /// Local cap that was exceeded.
143        max: u64,
144    },
145}
146
147impl PreambleError {
148    /// True if the error is fatal for the connection. Only
149    /// [`PreambleError::Incomplete`] is not.
150    pub const fn is_violation(self) -> bool {
151        !matches!(self, PreambleError::Incomplete)
152    }
153}
154
155impl fmt::Display for PreambleError {
156    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
157        match self {
158            PreambleError::Incomplete => f.write_str("incomplete preamble"),
159            PreambleError::BadMagic(b) => write!(f, "bad magic byte {b:#04x}, expected 0x57"),
160            PreambleError::UnknownKind(k) => write!(f, "unknown frame kind {k}"),
161            PreambleError::HeaderTooLarge { len, max } => {
162                write!(f, "header length {len} exceeds the limit of {max} bytes")
163            }
164        }
165    }
166}
167
168impl std::error::Error for PreambleError {}
169
170/// Parses a preamble from the front of `input`.
171///
172/// Returns the preamble and how many bytes it occupied. `max_header_bytes` is
173/// the local limit; a larger advertised length is rejected here, which is the
174/// only place where that check can happen before memory is committed.
175pub fn parse_preamble(
176    input: &[u8],
177    max_header_bytes: u64,
178) -> Result<(Preamble, usize), PreambleError> {
179    if input.len() < 2 {
180        return Err(PreambleError::Incomplete);
181    }
182    if input[0] != MAGIC {
183        return Err(PreambleError::BadMagic(input[0]));
184    }
185    let kind = FrameKind::from_u8(input[1]).ok_or(PreambleError::UnknownKind(input[1]))?;
186    let (header_len, used) = match decode_varint(&input[2..]) {
187        Ok(v) => v,
188        Err(VarintError::Truncated) => return Err(PreambleError::Incomplete),
189        // A varint cannot be out of range while decoding: two bits select the
190        // length, so every 8-byte form is within 2^62.
191        Err(VarintError::OutOfRange) => unreachable!("decoding cannot overflow"),
192    };
193    if header_len > max_header_bytes {
194        return Err(PreambleError::HeaderTooLarge {
195            len: header_len,
196            max: max_header_bytes,
197        });
198    }
199    Ok((Preamble { kind, header_len }, 2 + used))
200}
201
202/// A preamble for `header_len` bytes of header, and how many of the returned
203/// bytes are it.
204///
205/// Returned in a fixed array rather than pushed into a `Vec`, so a send path
206/// that already owns a buffer allocates nothing to write one: the frame's
207/// length field precedes the header it measures, so the header has to be
208/// encoded first, and a caller that encodes it at `MAX_PREAMBLE_LEN` can then
209/// right-align this preamble against it and write **one** contiguous slice.
210pub fn preamble_bytes(kind: FrameKind, header_len: u64) -> ([u8; MAX_PREAMBLE_LEN], usize) {
211    let mut bytes = [0u8; MAX_PREAMBLE_LEN];
212    bytes[0] = MAGIC;
213    bytes[1] = kind.to_u8();
214    let len = crate::varint::write_varint(header_len, &mut bytes[2..])
215        .expect("header lengths are bounded far below 2^62");
216    (bytes, 2 + len)
217}
218
219/// Appends a preamble for `header_len` bytes of header to `out`.
220pub fn encode_preamble(kind: FrameKind, header_len: u64, out: &mut Vec<u8>) {
221    let (bytes, len) = preamble_bytes(kind, header_len);
222    out.extend_from_slice(&bytes[..len]);
223}
224
225/// Builds a complete header-only frame: preamble followed by `header`.
226pub fn encode_frame(kind: FrameKind, header: &[u8]) -> Vec<u8> {
227    let mut out = Vec::with_capacity(MAX_PREAMBLE_LEN + header.len());
228    encode_preamble(kind, header.len() as u64, &mut out);
229    out.extend_from_slice(header);
230    out
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236
237    const CAP: u64 = 16 * 1024;
238
239    #[test]
240    fn kind_codes_match_the_protocol_document() {
241        let all = [
242            (FrameKind::Hello, 0u8),
243            (FrameKind::Data, 1),
244            (FrameKind::Error, 2),
245            (FrameKind::Subscribe, 3),
246            (FrameKind::Unsubscribe, 4),
247            (FrameKind::Credit, 5),
248            (FrameKind::Cursor, 6),
249            (FrameKind::Flow, 7),
250            (FrameKind::Report, 8),
251        ];
252        for (kind, code) in all {
253            assert_eq!(kind.to_u8(), code);
254            assert_eq!(FrameKind::from_u8(code), Some(kind));
255        }
256        for code in 9u8..=255 {
257            assert_eq!(FrameKind::from_u8(code), None, "kind {code}");
258        }
259    }
260
261    #[test]
262    fn only_data_carries_payload() {
263        assert!(FrameKind::Data.has_payload());
264        for k in [
265            FrameKind::Hello,
266            FrameKind::Error,
267            FrameKind::Subscribe,
268            FrameKind::Unsubscribe,
269            FrameKind::Credit,
270            FrameKind::Cursor,
271            FrameKind::Flow,
272            FrameKind::Report,
273        ] {
274            assert!(!k.has_payload(), "{k}");
275        }
276    }
277
278    #[test]
279    fn roundtrip_through_encode_and_parse() {
280        for len in [0u64, 1, 63, 64, 16_383, 16_384] {
281            let mut buf = Vec::new();
282            encode_preamble(FrameKind::Data, len, &mut buf);
283            let (p, used) = parse_preamble(&buf, CAP).unwrap();
284            assert_eq!(p.kind, FrameKind::Data);
285            assert_eq!(p.header_len, len);
286            assert_eq!(used, buf.len());
287        }
288    }
289
290    #[test]
291    fn every_kind_encodes_its_documented_kind_byte() {
292        // Per-kind preamble encoding and round-trip. The golden *frames* of
293        // docs/PROTOCOL.md §8 are asserted end-to-end, against real headers,
294        // in `tests/golden_vectors.rs`; this covers the kind byte for every
295        // kind, including those no golden vector exercises.
296        for (kind, code) in [
297            (FrameKind::Hello, 0x00u8),
298            (FrameKind::Data, 0x01),
299            (FrameKind::Error, 0x02),
300            (FrameKind::Subscribe, 0x03),
301            (FrameKind::Unsubscribe, 0x04),
302            (FrameKind::Credit, 0x05),
303            (FrameKind::Cursor, 0x06),
304            (FrameKind::Flow, 0x07),
305            (FrameKind::Report, 0x08),
306        ] {
307            for header_len in [0usize, 3, 5, 9, 11, 16, 18] {
308                let frame = encode_frame(kind, &vec![0; header_len]);
309                assert_eq!(frame[0], MAGIC, "{kind}: magic");
310                assert_eq!(frame[1], code, "{kind}: kind byte");
311                let (preamble, used) = parse_preamble(&frame, CAP).unwrap();
312                assert_eq!(preamble.kind, kind);
313                assert_eq!(preamble.header_len as usize, header_len);
314                assert_eq!(frame.len() - used, header_len, "{kind}: header follows");
315            }
316        }
317    }
318
319    #[test]
320    fn incomplete_input_is_not_a_violation() {
321        for prefix in [
322            &[][..],
323            &[MAGIC][..],
324            &[MAGIC, 1, 0x80][..],
325            &[MAGIC, 1, 0xc0, 0, 0][..],
326        ] {
327            let err = parse_preamble(prefix, CAP).unwrap_err();
328            assert_eq!(err, PreambleError::Incomplete, "{prefix:?}");
329            assert!(!err.is_violation());
330        }
331    }
332
333    #[test]
334    fn bad_magic_is_a_violation() {
335        let err = parse_preamble(&[0x58, 0x01, 0x00], CAP).unwrap_err();
336        assert_eq!(err, PreambleError::BadMagic(0x58));
337        assert!(err.is_violation());
338    }
339
340    #[test]
341    fn unknown_kind_is_a_violation() {
342        // Kind `9` is the first free one: `5` is CREDIT
343        // (`docs/decisions/0018-minimal-broker.md` §4.6), `6` is CURSOR
344        // (`docs/decisions/0024-three-families-one-back-channel.md` §4.4),
345        // `7` is FLOW (`docs/decisions/0034-late-is-lost.md` §4.2) and
346        // `8` is REPORT (`docs/decisions/0036-connection-statistics.md`
347        // §4.5).
348        let err = parse_preamble(&[MAGIC, 0x09, 0x00], CAP).unwrap_err();
349        assert_eq!(err, PreambleError::UnknownKind(9));
350        assert!(err.is_violation());
351    }
352
353    #[test]
354    fn oversized_header_is_rejected_before_allocation() {
355        // 1 MiB advertised against a 16 KiB cap.
356        let mut buf = vec![MAGIC, FrameKind::Data.to_u8()];
357        crate::varint::encode_varint(1024 * 1024, &mut buf).unwrap();
358        let err = parse_preamble(&buf, CAP).unwrap_err();
359        assert_eq!(
360            err,
361            PreambleError::HeaderTooLarge {
362                len: 1024 * 1024,
363                max: CAP
364            }
365        );
366        assert!(err.is_violation());
367    }
368
369    #[test]
370    fn a_header_exactly_at_the_cap_is_accepted() {
371        let mut buf = vec![MAGIC, FrameKind::Hello.to_u8()];
372        crate::varint::encode_varint(CAP, &mut buf).unwrap();
373        assert_eq!(parse_preamble(&buf, CAP).unwrap().0.header_len, CAP);
374    }
375
376    #[test]
377    fn non_minimal_length_encodings_are_accepted() {
378        // header_len = 5 in the 8-byte form.
379        let buf = [MAGIC, FrameKind::Error.to_u8(), 0xc0, 0, 0, 0, 0, 0, 0, 5];
380        let (p, used) = parse_preamble(&buf, CAP).unwrap();
381        assert_eq!(p.header_len, 5);
382        assert_eq!(used, 10);
383    }
384
385    #[test]
386    fn payload_bytes_after_the_preamble_are_untouched() {
387        let frame = encode_frame(FrameKind::Data, &[0xaa, 0xbb]);
388        let (p, used) = parse_preamble(&frame, CAP).unwrap();
389        assert_eq!(p.header_len, 2);
390        assert_eq!(&frame[used..], &[0xaa, 0xbb]);
391    }
392}
weida_protocol/frame.rs

weida_protocol/
frame.rs
