cu29_logstream/
wire.rs

1//! Packed transport framing. No version or schema is sent in packet headers;
2//! sender and receiver must be built for the matching application.
3
4use crate::{Error, RecordKind, Result};
5use alloc::vec::Vec;
6
7const PACKET_MAGIC: [u8; 4] = *b"CULS";
8/// Maximum packet header length, used to bound storage and the shared FEC symbol size.
9/// RaptorQ uses 51 bytes; RLC uses 32 for source packets and 36 for repair packets.
10pub const PACKET_HEADER_LEN: usize = 51;
11const COMMON_HEADER_LEN: usize = 28;
12const RLC_SOURCE_HEADER_LEN: usize = 32;
13const RLC_REPAIR_HEADER_LEN: usize = 36;
14
15const fn packet_header_len(scheme: FecScheme, kind: FecSymbolKind) -> usize {
16    match (scheme, kind) {
17        (FecScheme::RaptorQ, _) => PACKET_HEADER_LEN,
18        (_, FecSymbolKind::Source) => RLC_SOURCE_HEADER_LEN,
19        (_, FecSymbolKind::Repair) => RLC_REPAIR_HEADER_LEN,
20    }
21}
22
23/// Independently scheduled transport lanes.
24#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25#[repr(u8)]
26pub enum Lane {
27    Control = 0,
28    ReplayCritical = 1,
29    LargeObject = 2,
30    StructuredLog = 3,
31    Visualization = 4,
32}
33
34impl TryFrom<u8> for Lane {
35    type Error = Error;
36
37    fn try_from(value: u8) -> Result<Self> {
38        match value {
39            0 => Ok(Self::Control),
40            1 => Ok(Self::ReplayCritical),
41            2 => Ok(Self::LargeObject),
42            3 => Ok(Self::StructuredLog),
43            4 => Ok(Self::Visualization),
44            _ => Err(Error::UnknownLane(value)),
45        }
46    }
47}
48
49#[derive(Clone, Copy, Debug, PartialEq, Eq)]
50#[repr(u8)]
51pub enum FecScheme {
52    RlcGf2 = 0,
53    RlcGf256 = 1,
54    RaptorQ = 2,
55}
56
57impl TryFrom<u8> for FecScheme {
58    type Error = Error;
59
60    fn try_from(value: u8) -> Result<Self> {
61        match value {
62            0 => Ok(Self::RlcGf2),
63            1 => Ok(Self::RlcGf256),
64            2 => Ok(Self::RaptorQ),
65            _ => Err(Error::UnknownFecScheme(value)),
66        }
67    }
68}
69
70#[derive(Clone, Copy, Debug, PartialEq, Eq)]
71#[repr(u8)]
72pub enum FecSymbolKind {
73    Source = 0,
74    Repair = 1,
75}
76
77impl TryFrom<u8> for FecSymbolKind {
78    type Error = Error;
79
80    fn try_from(value: u8) -> Result<Self> {
81        match value {
82            0 => Ok(Self::Source),
83            1 => Ok(Self::Repair),
84            _ => Err(Error::UnknownSymbolKind(value)),
85        }
86    }
87}
88
89/// Fixed-endian metadata repeated on every independently reorderable symbol.
90#[derive(Clone, Copy, Debug, PartialEq, Eq)]
91pub struct WireHeader {
92    pub lane: Lane,
93    pub record_kind: RecordKind,
94    pub fec_scheme: FecScheme,
95    pub symbol_kind: FecSymbolKind,
96    pub session_id: [u8; 16],
97    pub sender_id: u32,
98    /// RaptorQ object identity. Not transmitted for RLC; decoded as zero.
99    pub object_id: u64,
100    /// RaptorQ OTI (12 bytes), RLC source ID (first 4), or RLC repair ID (first 8).
101    /// RaptorQ's reserved byte at index 5 must be zero; it is omitted on the wire
102    /// and restored locally when decoding.
103    /// Unused RLC bytes are not transmitted and decode as zero.
104    pub fec_metadata: [u8; 12],
105    /// RaptorQ payload ID. Not transmitted for RLC; decoded as zero.
106    pub fragment_count: u32,
107}
108
109/// Owned packet for callers that retain or edit payloads.
110/// Receiver parsing uses [`WirePacketRef`] to borrow the datagram instead.
111#[derive(Clone, Debug, PartialEq, Eq)]
112pub struct WirePacket {
113    pub header: WireHeader,
114    pub payload: Vec<u8>,
115}
116
117impl WirePacket {
118    pub fn encode(&self) -> Result<Vec<u8>> {
119        let mut bytes = alloc::vec![0; packet_header_len(self.header.fec_scheme, self.header.symbol_kind) + self.payload.len()];
120        let encoded = encode_packet_into(self.header, &self.payload, &mut bytes)?;
121        debug_assert_eq!(encoded, bytes.len());
122        Ok(bytes)
123    }
124
125    pub fn decode(bytes: &[u8]) -> Result<Self> {
126        let packet = WirePacketRef::decode(bytes)?;
127        Ok(Self {
128            header: packet.header,
129            payload: packet.payload.to_vec(),
130        })
131    }
132}
133
134/// Structurally validated packet view borrowing the transport's receive buffer.
135#[derive(Clone, Copy, Debug, PartialEq, Eq)]
136pub struct WirePacketRef<'a> {
137    pub header: WireHeader,
138    pub payload: &'a [u8],
139}
140
141impl<'a> WirePacketRef<'a> {
142    /// Decodes exactly one complete packet supplied by the carrier.
143    /// The carrier must verify packet integrity before calling this parser.
144    /// This validates header structure only; FEC cannot correct corrupted symbols.
145    /// The payload occupies the remaining packet extent after the header.
146    pub fn decode(bytes: &'a [u8]) -> Result<Self> {
147        if bytes.len() < COMMON_HEADER_LEN {
148            return Err(Error::TruncatedPacket);
149        }
150        if bytes[..4] != PACKET_MAGIC {
151            return Err(Error::InvalidMagic);
152        }
153        let fec_scheme = FecScheme::try_from(bytes[6])?;
154        let symbol_kind = FecSymbolKind::try_from(bytes[7])?;
155        let header_len = packet_header_len(fec_scheme, symbol_kind);
156        if bytes.len() < header_len {
157            return Err(Error::TruncatedPacket);
158        }
159        let mut session_id = [0_u8; 16];
160        session_id.copy_from_slice(&bytes[8..24]);
161        let mut fec_metadata = [0_u8; 12];
162        let (object_id, fragment_count) = match fec_scheme {
163            FecScheme::RaptorQ => {
164                fec_metadata[..5].copy_from_slice(&bytes[36..41]);
165                fec_metadata[6..].copy_from_slice(&bytes[41..47]);
166                (
167                    u64::from_be_bytes(bytes[28..36].try_into().unwrap()),
168                    u32::from_be_bytes(bytes[47..51].try_into().unwrap()),
169                )
170            }
171            FecScheme::RlcGf2 | FecScheme::RlcGf256 => {
172                let metadata_len = header_len - COMMON_HEADER_LEN;
173                fec_metadata[..metadata_len].copy_from_slice(&bytes[COMMON_HEADER_LEN..header_len]);
174                (0, 0)
175            }
176        };
177
178        Ok(Self {
179            header: WireHeader {
180                lane: Lane::try_from(bytes[4])?,
181                record_kind: RecordKind::try_from(bytes[5])?,
182                fec_scheme,
183                symbol_kind,
184                session_id,
185                sender_id: u32::from_be_bytes(bytes[24..28].try_into().unwrap()),
186                object_id,
187                fec_metadata,
188                fragment_count,
189            },
190            payload: &bytes[header_len..],
191        })
192    }
193}
194
195/// Encodes one packet into caller-owned storage and returns its exact length.
196pub fn encode_packet_into(header: WireHeader, payload: &[u8], output: &mut [u8]) -> Result<usize> {
197    if header.fec_scheme == FecScheme::RaptorQ && header.fec_metadata[5] != 0 {
198        return Err(Error::InvalidFecMetadata(
199            "RaptorQ OTI reserved byte is nonzero",
200        ));
201    }
202    let header_len = packet_header_len(header.fec_scheme, header.symbol_kind);
203    let needed = header_len
204        .checked_add(payload.len())
205        .ok_or(Error::InvalidConfig("packet length overflow"))?;
206    if output.len() < needed {
207        return Err(Error::BufferTooSmall {
208            needed,
209            available: output.len(),
210        });
211    }
212
213    let bytes = &mut output[..needed];
214    bytes.fill(0);
215    bytes[..4].copy_from_slice(&PACKET_MAGIC);
216    bytes[4] = header.lane as u8;
217    bytes[5] = header.record_kind as u8;
218    bytes[6] = header.fec_scheme as u8;
219    bytes[7] = header.symbol_kind as u8;
220    bytes[8..24].copy_from_slice(&header.session_id);
221    bytes[24..28].copy_from_slice(&header.sender_id.to_be_bytes());
222    match header.fec_scheme {
223        FecScheme::RaptorQ => {
224            bytes[28..36].copy_from_slice(&header.object_id.to_be_bytes());
225            bytes[36..41].copy_from_slice(&header.fec_metadata[..5]);
226            bytes[41..47].copy_from_slice(&header.fec_metadata[6..]);
227            bytes[47..51].copy_from_slice(&header.fragment_count.to_be_bytes());
228        }
229        FecScheme::RlcGf2 | FecScheme::RlcGf256 => {
230            bytes[COMMON_HEADER_LEN..header_len]
231                .copy_from_slice(&header.fec_metadata[..header_len - COMMON_HEADER_LEN]);
232        }
233    }
234    bytes[header_len..].copy_from_slice(payload);
235    Ok(needed)
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241    use alloc::vec;
242
243    fn fixture() -> WirePacket {
244        WirePacket {
245            header: WireHeader {
246                lane: Lane::LargeObject,
247                record_kind: RecordKind::KeyFrame,
248                fec_scheme: FecScheme::RaptorQ,
249                symbol_kind: FecSymbolKind::Source,
250                session_id: [0x11; 16],
251                sender_id: 0x2233_4455,
252                object_id: 0x0102_0304_0506_0708,
253                fec_metadata: [0, 0, 0, 0, 3, 0, 0, 8, 1, 0, 1, 1],
254                fragment_count: 1,
255            },
256            payload: vec![0, 0, 0, 0, 1, 2, 3, 0, 0, 0, 0, 0],
257        }
258    }
259
260    #[test]
261    fn fixed_header_has_a_stable_golden_prefix() {
262        let encoded = fixture().encode().unwrap();
263        assert_eq!(&encoded[..8], &[b'C', b'U', b'L', b'S', 2, 2, 2, 0]);
264        assert_eq!(encoded.len(), 51 + fixture().payload.len());
265        assert_eq!(&encoded[8..24], &[0x11; 16]);
266        assert_eq!(&encoded[24..28], &0x2233_4455_u32.to_be_bytes());
267        assert_eq!(&encoded[28..36], &0x0102_0304_0506_0708_u64.to_be_bytes());
268        assert_eq!(&encoded[36..47], &[0, 0, 0, 0, 3, 0, 8, 1, 0, 1, 1]);
269        assert_eq!(&encoded[47..51], &1_u32.to_be_bytes());
270        assert_eq!(&encoded[51..], fixture().payload);
271        assert_eq!(WirePacket::decode(&encoded).unwrap(), fixture());
272    }
273
274    #[test]
275    fn raptorq_oti_preserves_every_nonreserved_byte() {
276        for kind in [FecSymbolKind::Source, FecSymbolKind::Repair] {
277            let mut packet = fixture();
278            packet.header.symbol_kind = kind;
279            packet.header.fec_metadata = [1, 2, 3, 4, 5, 0, 6, 7, 8, 9, 10, 11];
280            packet.header.fragment_count = 0x1234_5678;
281            let encoded = packet.encode().unwrap();
282            assert_eq!(&encoded[36..47], &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
283            assert_eq!(&encoded[47..51], &[0x12, 0x34, 0x56, 0x78]);
284            assert_eq!(WirePacket::decode(&encoded).unwrap(), packet);
285        }
286    }
287
288    #[test]
289    fn raptorq_oti_rejects_nonzero_reserved_byte_before_encoding() {
290        for kind in [FecSymbolKind::Source, FecSymbolKind::Repair] {
291            let mut packet = fixture();
292            packet.header.symbol_kind = kind;
293            for reserved in 1..=u8::MAX {
294                packet.header.fec_metadata[5] = reserved;
295                let expected = Error::InvalidFecMetadata("RaptorQ OTI reserved byte is nonzero");
296                let mut storage = [0xaa; 128];
297                assert_eq!(
298                    encode_packet_into(packet.header, &packet.payload, &mut storage),
299                    Err(expected.clone())
300                );
301                assert_eq!(storage, [0xaa; 128]);
302                assert_eq!(packet.encode(), Err(expected));
303            }
304        }
305    }
306
307    fn rlc_fixture(scheme: FecScheme, kind: FecSymbolKind) -> WirePacket {
308        let mut packet = fixture();
309        packet.header.lane = Lane::ReplayCritical;
310        packet.header.record_kind = RecordKind::CopperList;
311        packet.header.fec_scheme = scheme;
312        packet.header.symbol_kind = kind;
313        packet.header.object_id = 0;
314        packet.header.fragment_count = 0;
315        packet.header.fec_metadata = [0; 12];
316        let id_len = match kind {
317            FecSymbolKind::Source => 4,
318            FecSymbolKind::Repair => 8,
319        };
320        packet.header.fec_metadata[..id_len].copy_from_slice(&[1, 2, 3, 4, 5, 6, 7, 8][..id_len]);
321        packet
322    }
323
324    #[test]
325    fn rlc_headers_carry_only_the_active_fec_id() {
326        for scheme in [FecScheme::RlcGf2, FecScheme::RlcGf256] {
327            for (kind, header_len, id_len) in [
328                (FecSymbolKind::Source, 32, 4),
329                (FecSymbolKind::Repair, 36, 8),
330            ] {
331                let packet = rlc_fixture(scheme, kind);
332                let encoded = packet.encode().unwrap();
333                assert_eq!(encoded.len(), header_len + packet.payload.len());
334                assert_eq!(
335                    &encoded[..8],
336                    &[
337                        b'C',
338                        b'U',
339                        b'L',
340                        b'S',
341                        1,
342                        RecordKind::CopperList as u8,
343                        scheme as u8,
344                        kind as u8
345                    ]
346                );
347                assert_eq!(&encoded[8..24], &[0x11; 16]);
348                assert_eq!(&encoded[24..28], &0x2233_4455_u32.to_be_bytes());
349                assert_eq!(
350                    &encoded[28..28 + id_len],
351                    &packet.header.fec_metadata[..id_len]
352                );
353                assert_eq!(header_len, 28 + id_len);
354                assert_eq!(&encoded[header_len..], packet.payload);
355                assert_eq!(WirePacket::decode(&encoded).unwrap(), packet);
356
357                let mut exact_storage = vec![0; encoded.len()];
358                assert_eq!(
359                    encode_packet_into(packet.header, &packet.payload, &mut exact_storage).unwrap(),
360                    encoded.len()
361                );
362                assert_eq!(exact_storage, encoded);
363                let available = encoded.len() - 1;
364                assert_eq!(
365                    encode_packet_into(
366                        packet.header,
367                        &packet.payload,
368                        &mut exact_storage[..available]
369                    ),
370                    Err(Error::BufferTooSmall {
371                        needed: encoded.len(),
372                        available
373                    })
374                );
375            }
376        }
377    }
378
379    #[test]
380    fn all_header_layouts_reject_truncated_headers_and_invalid_tags() {
381        for scheme in [FecScheme::RlcGf2, FecScheme::RlcGf256, FecScheme::RaptorQ] {
382            for kind in [FecSymbolKind::Source, FecSymbolKind::Repair] {
383                let mut packet = if scheme == FecScheme::RaptorQ {
384                    fixture()
385                } else {
386                    rlc_fixture(scheme, kind)
387                };
388                packet.header.symbol_kind = kind;
389                for payload in [vec![], vec![0x5a; 64]] {
390                    packet.payload = payload;
391                    let encoded = packet.encode().unwrap();
392                    assert_eq!(WirePacket::decode(&encoded).unwrap(), packet);
393                    let header_len = packet_header_len(scheme, kind);
394                    for end in 0..header_len {
395                        assert_eq!(
396                            WirePacketRef::decode(&encoded[..end]),
397                            Err(Error::TruncatedPacket)
398                        );
399                    }
400                    for (offset, expected) in [
401                        (0, Error::InvalidMagic),
402                        (4, Error::UnknownLane(255)),
403                        (5, Error::UnknownRecordKind(255)),
404                        (6, Error::UnknownFecScheme(255)),
405                        (7, Error::UnknownSymbolKind(255)),
406                    ] {
407                        let mut invalid = encoded.clone();
408                        invalid[offset] = 255;
409                        assert_eq!(WirePacketRef::decode(&invalid), Err(expected));
410                    }
411                }
412            }
413        }
414    }
415
416    #[test]
417    fn packet_extent_defines_payload_length_for_every_layout() {
418        for scheme in [FecScheme::RlcGf2, FecScheme::RlcGf256, FecScheme::RaptorQ] {
419            for kind in [FecSymbolKind::Source, FecSymbolKind::Repair] {
420                let mut packet = if scheme == FecScheme::RaptorQ {
421                    fixture()
422                } else {
423                    rlc_fixture(scheme, kind)
424                };
425                packet.header.symbol_kind = kind;
426                let header_len = packet_header_len(scheme, kind);
427                for payload_len in [0, 1, crate::DEFAULT_MAX_SYMBOL_SIZE, u16::MAX as usize + 1] {
428                    packet.payload = vec![0x5a; payload_len];
429                    let needed = header_len + payload_len;
430                    let mut storage = vec![0xaa; needed + 1];
431                    assert_eq!(
432                        encode_packet_into(packet.header, &packet.payload, &mut storage).unwrap(),
433                        needed
434                    );
435                    assert_eq!(storage[needed], 0xaa);
436                    let decoded = WirePacketRef::decode(&storage[..needed]).unwrap();
437                    assert_eq!(decoded.header, packet.header);
438                    assert_eq!(decoded.payload, packet.payload);
439                    assert_eq!(decoded.payload.as_ptr(), storage[header_len..].as_ptr());
440                    // Complete packet boundaries and integrity belong to the carrier.
441                    let extended = WirePacketRef::decode(&storage).unwrap();
442                    assert_eq!(extended.payload.len(), payload_len + 1);
443                    assert_eq!(extended.payload[payload_len], 0xaa);
444                    for end in [header_len, header_len + payload_len / 2, needed] {
445                        assert_eq!(
446                            WirePacketRef::decode(&storage[..end]).unwrap().payload,
447                            &packet.payload[..end - header_len]
448                        );
449                    }
450                    assert_eq!(
451                        encode_packet_into(
452                            packet.header,
453                            &packet.payload,
454                            &mut storage[..needed - 1]
455                        ),
456                        Err(Error::BufferTooSmall {
457                            needed,
458                            available: needed - 1,
459                        })
460                    );
461                }
462            }
463        }
464    }
465
466    #[test]
467    fn caller_owned_encoding_matches_allocating_encoding() {
468        let packet = fixture();
469        let expected = packet.encode().unwrap();
470        let mut storage = [0xaa; 128];
471        let encoded = encode_packet_into(packet.header, &packet.payload, &mut storage).unwrap();
472
473        assert_eq!(&storage[..encoded], expected);
474        assert!(storage[encoded..].iter().all(|byte| *byte == 0xaa));
475    }
476}
cu29_logstream/wire.rs

cu29_logstream/
wire.rs
