cu29_logstream/
object.rs

1//! Bounded finite-object transport using systematic RFC 6330 RaptorQ.
2
3use crate::{
4    Error, FecScheme, FecSymbolKind, Lane, PACKET_HEADER_LEN, ReceiveError, RecordKind,
5    RecoveredRecord, Result, StreamIdentity, WireHeader, WirePacketRef, decode_record,
6    encode_packet_into,
7};
8use alloc::{vec, vec::Vec};
9use raptorq::{Decoder, Encoder, EncodingPacket, ObjectTransmissionInformation, PayloadId};
10
11const RFC6330_MAX_TRANSFER_LENGTH: u64 = 942_574_504_275;
12const RFC6330_MAX_SOURCE_SYMBOLS_PER_BLOCK: u64 = 56_403;
13
14/// Sender policy for one finite-object lane.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub struct FiniteObjectSenderConfig {
17    pub identity: StreamIdentity,
18    pub lane: Lane,
19    /// RaptorQ symbol bytes, excluding the Copper wire header.
20    pub symbol_size: u16,
21    /// Hard limit checked before constructing the RaptorQ encoder.
22    pub max_object_bytes: u64,
23    /// Repair symbols emitted for every RaptorQ source block.
24    pub repair_symbols_per_block: u32,
25}
26
27/// Receiver-local limits checked before any RaptorQ decoder is constructed.
28#[derive(Clone, Copy, Debug, PartialEq, Eq)]
29pub struct FiniteObjectLimits {
30    pub max_object_bytes: u64,
31    pub max_symbol_size: u16,
32    pub max_concurrent_objects: usize,
33}
34
35impl FiniteObjectLimits {
36    pub const fn new(
37        max_object_bytes: u64,
38        max_symbol_size: u16,
39        max_concurrent_objects: usize,
40    ) -> Self {
41        Self {
42            max_object_bytes,
43            max_symbol_size,
44            max_concurrent_objects,
45        }
46    }
47}
48
49/// Observed finite-object receive behavior.
50#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
51pub struct FiniteObjectRecoveryStats {
52    pub datagrams_seen: usize,
53    pub valid_datagrams: usize,
54    pub invalid_datagrams: usize,
55    pub inconsistent_datagrams: usize,
56    pub duplicate_symbols: usize,
57    pub source_symbols_received: usize,
58    pub repair_symbols_received: usize,
59    pub objects_recovered: usize,
60    pub peak_concurrent_objects: usize,
61}
62
63/// RaptorQ encoder for immutable framed semantic records.
64///
65/// Construction and encoding allocate and must run on a non-real-time output worker.
66pub struct FiniteObjectEncoder {
67    config: FiniteObjectSenderConfig,
68}
69
70impl FiniteObjectEncoder {
71    pub fn new(config: FiniteObjectSenderConfig) -> Result<Self> {
72        validate_sender_config(config)?;
73        Ok(Self { config })
74    }
75
76    pub const fn config(&self) -> FiniteObjectSenderConfig {
77        self.config
78    }
79
80    /// Encodes one already-framed semantic record and appends its datagrams.
81    pub fn push_record(&mut self, record: &[u8], output: &mut Vec<Vec<u8>>) -> Result<usize> {
82        self.push_record_with(record, |datagram| {
83            output.push(datagram.to_vec());
84            Ok(())
85        })
86    }
87
88    /// Encodes one record and emits each datagram immediately.
89    pub fn push_record_with(
90        &mut self,
91        record: &[u8],
92        mut emit: impl FnMut(&[u8]) -> Result<()>,
93    ) -> Result<usize> {
94        let decoded = decode_record(record)?;
95        let record_len = u64::try_from(record.len())
96            .map_err(|_| Error::InvalidConfig("finite object length exceeds u64"))?;
97        if record_len > self.config.max_object_bytes {
98            return Err(Error::ObjectTooLarge {
99                actual: record_len,
100                maximum: self.config.max_object_bytes,
101            });
102        }
103
104        let encoder = Encoder::with_defaults(record, self.config.symbol_size);
105        let oti = encoder.get_config().serialize();
106        let datagram_len = PACKET_HEADER_LEN
107            .checked_add(self.config.symbol_size as usize)
108            .ok_or(Error::InvalidConfig(
109                "finite-object datagram length overflow",
110            ))?;
111        let mut datagram = vec![0; datagram_len];
112        let mut emitted = 0_usize;
113
114        for block in encoder.get_block_encoders() {
115            for packet in block.source_packets() {
116                self.emit_packet(
117                    decoded,
118                    FecSymbolKind::Source,
119                    oti,
120                    packet,
121                    &mut datagram,
122                    &mut emit,
123                )?;
124                emitted = emitted.saturating_add(1);
125            }
126            for packet in block.repair_packets(0, self.config.repair_symbols_per_block) {
127                self.emit_packet(
128                    decoded,
129                    FecSymbolKind::Repair,
130                    oti,
131                    packet,
132                    &mut datagram,
133                    &mut emit,
134                )?;
135                emitted = emitted.saturating_add(1);
136            }
137        }
138        Ok(emitted)
139    }
140
141    fn emit_packet(
142        &mut self,
143        record: crate::DecodedRecord<'_>,
144        symbol_kind: FecSymbolKind,
145        oti: [u8; 12],
146        packet: EncodingPacket,
147        datagram: &mut [u8],
148        emit: &mut impl FnMut(&[u8]) -> Result<()>,
149    ) -> Result<()> {
150        let (payload_id, payload) = packet.split();
151        let header = WireHeader {
152            lane: self.config.lane,
153            record_kind: record.kind,
154            fec_scheme: FecScheme::RaptorQ,
155            symbol_kind,
156            session_id: self.config.identity.session_id,
157            sender_id: self.config.identity.sender_id,
158            object_id: record.object_id,
159            fec_metadata: oti,
160            fragment_count: u32::from_be_bytes(payload_id.serialize()),
161        };
162        let encoded = encode_packet_into(header, &payload, datagram)?;
163        emit(&datagram[..encoded])?;
164        Ok(())
165    }
166}
167
168/// Bounded assembler for independently reorderable finite RaptorQ objects.
169pub struct FiniteObjectDecoder {
170    identity: StreamIdentity,
171    lane: Lane,
172    limits: FiniteObjectLimits,
173    objects: Vec<ObjectDecoder>,
174    ready: Vec<ReadyObject>,
175    completed: Vec<CompletedObject>,
176    stats: FiniteObjectRecoveryStats,
177}
178
179impl FiniteObjectDecoder {
180    pub fn new(identity: StreamIdentity, lane: Lane, limits: FiniteObjectLimits) -> Result<Self> {
181        validate_limits(limits)?;
182        Ok(Self {
183            identity,
184            lane,
185            limits,
186            objects: Vec::with_capacity(limits.max_concurrent_objects),
187            ready: Vec::with_capacity(limits.max_concurrent_objects),
188            completed: Vec::with_capacity(limits.max_concurrent_objects),
189            stats: FiniteObjectRecoveryStats::default(),
190        })
191    }
192
193    pub const fn stats(&self) -> FiniteObjectRecoveryStats {
194        self.stats
195    }
196
197    pub fn occupancy(&self) -> usize {
198        self.objects.len().saturating_add(self.ready.len())
199    }
200
201    /// Accepts one carrier-verified datagram and emits a digest-verified record
202    /// when its object completes. Callers must uphold [`crate::CuStreamRx`] integrity.
203    pub fn receive_datagram<E>(
204        &mut self,
205        datagram: &[u8],
206        mut emit: impl FnMut(&RecoveredRecord) -> core::result::Result<(), E>,
207    ) -> core::result::Result<(), ReceiveError<E>> {
208        self.drain_records(&mut emit)?;
209        let packet = match WirePacketRef::decode(datagram) {
210            Ok(packet) => packet,
211            Err(_) => {
212                self.stats.datagrams_seen = self.stats.datagrams_seen.saturating_add(1);
213                self.stats.invalid_datagrams = self.stats.invalid_datagrams.saturating_add(1);
214                return Ok(());
215            }
216        };
217        self.receive_packet(packet)?;
218        self.drain_records(&mut emit)
219    }
220
221    pub(crate) fn receive_packet(&mut self, packet: WirePacketRef<'_>) -> Result<()> {
222        self.stats.datagrams_seen = self.stats.datagrams_seen.saturating_add(1);
223        if packet.header.session_id != self.identity.session_id
224            || packet.header.sender_id != self.identity.sender_id
225            || packet.header.lane != self.lane
226            || packet.header.fec_scheme != FecScheme::RaptorQ
227            || packet.header.record_kind == RecordKind::CopperList
228        {
229            self.stats.inconsistent_datagrams = self.stats.inconsistent_datagrams.saturating_add(1);
230            return Ok(());
231        }
232
233        let oti = match validate_oti(packet.header.fec_metadata, self.limits) {
234            Ok(oti) => oti,
235            Err(_) => {
236                self.stats.invalid_datagrams = self.stats.invalid_datagrams.saturating_add(1);
237                return Ok(());
238            }
239        };
240        if packet.payload.len() != oti.symbol_size() as usize {
241            self.stats.invalid_datagrams = self.stats.invalid_datagrams.saturating_add(1);
242            return Ok(());
243        }
244
245        let payload_id = PayloadId::deserialize(&packet.header.fragment_count.to_be_bytes());
246        if payload_id.source_block_number() >= oti.source_blocks() {
247            self.stats.invalid_datagrams = self.stats.invalid_datagrams.saturating_add(1);
248            return Ok(());
249        }
250        if let Some(object) = self.ready.iter().find(|object| {
251            object.kind == packet.header.record_kind && object.object_id == packet.header.object_id
252        }) {
253            if object.oti != oti {
254                self.stats.inconsistent_datagrams =
255                    self.stats.inconsistent_datagrams.saturating_add(1);
256            } else {
257                self.stats.duplicate_symbols = self.stats.duplicate_symbols.saturating_add(1);
258            }
259            return Ok(());
260        }
261        if let Some(object) = self.completed.iter().find(|object| {
262            object.kind == packet.header.record_kind && object.object_id == packet.header.object_id
263        }) {
264            if object.oti != oti {
265                self.stats.inconsistent_datagrams =
266                    self.stats.inconsistent_datagrams.saturating_add(1);
267            } else {
268                self.stats.duplicate_symbols = self.stats.duplicate_symbols.saturating_add(1);
269            }
270            return Ok(());
271        }
272        let object_index = match self.objects.iter().position(|object| {
273            object.kind == packet.header.record_kind && object.object_id == packet.header.object_id
274        }) {
275            Some(index) => {
276                let object = &self.objects[index];
277                if object.oti != oti {
278                    self.stats.inconsistent_datagrams =
279                        self.stats.inconsistent_datagrams.saturating_add(1);
280                    return Ok(());
281                }
282                index
283            }
284            None => {
285                let actual = self
286                    .objects
287                    .len()
288                    .saturating_add(self.ready.len())
289                    .saturating_add(1);
290                if actual > self.limits.max_concurrent_objects {
291                    return Err(Error::TooManyRecords {
292                        actual,
293                        maximum: self.limits.max_concurrent_objects,
294                    });
295                }
296                self.objects.push(ObjectDecoder {
297                    kind: packet.header.record_kind,
298                    object_id: packet.header.object_id,
299                    oti,
300                    decoder: Decoder::new(oti),
301                });
302                self.stats.peak_concurrent_objects =
303                    self.stats.peak_concurrent_objects.max(self.objects.len());
304                self.objects.len() - 1
305            }
306        };
307
308        match packet.header.symbol_kind {
309            FecSymbolKind::Source => {
310                self.stats.source_symbols_received =
311                    self.stats.source_symbols_received.saturating_add(1);
312            }
313            FecSymbolKind::Repair => {
314                self.stats.repair_symbols_received =
315                    self.stats.repair_symbols_received.saturating_add(1);
316            }
317        }
318        self.stats.valid_datagrams = self.stats.valid_datagrams.saturating_add(1);
319        let completed = self.objects[object_index]
320            .decoder
321            .decode(EncodingPacket::new(payload_id, packet.payload.to_vec()));
322        let Some(bytes) = completed else {
323            return Ok(());
324        };
325
326        let object = self.objects.swap_remove(object_index);
327        let record = RecoveredRecord::from_bytes(bytes)?;
328        let decoded = record.decoded();
329        if decoded.kind != object.kind || decoded.object_id != object.object_id {
330            return Err(Error::InconsistentObject);
331        }
332        self.ready.push(ReadyObject {
333            kind: object.kind,
334            object_id: object.object_id,
335            oti: object.oti,
336            record,
337        });
338        self.stats.objects_recovered = self.stats.objects_recovered.saturating_add(1);
339        Ok(())
340    }
341
342    /// Retries delivery of records retained after a consumer failure.
343    pub fn drain_records<E>(
344        &mut self,
345        mut emit: impl FnMut(&RecoveredRecord) -> core::result::Result<(), E>,
346    ) -> core::result::Result<(), ReceiveError<E>> {
347        while let Some(ready) = self.ready.first() {
348            emit(&ready.record).map_err(ReceiveError::Consumer)?;
349            self.pop_record();
350        }
351        Ok(())
352    }
353
354    pub(crate) fn pop_record(&mut self) -> Option<RecoveredRecord> {
355        if self.ready.is_empty() {
356            return None;
357        }
358        let delivered = self.ready.remove(0);
359        if self.completed.len() == self.limits.max_concurrent_objects {
360            self.completed.remove(0);
361        }
362        self.completed.push(CompletedObject {
363            kind: delivered.kind,
364            object_id: delivered.object_id,
365            oti: delivered.oti,
366        });
367        Some(delivered.record)
368    }
369}
370
371struct ObjectDecoder {
372    kind: RecordKind,
373    object_id: u64,
374    oti: ObjectTransmissionInformation,
375    decoder: Decoder,
376}
377
378struct ReadyObject {
379    kind: RecordKind,
380    object_id: u64,
381    oti: ObjectTransmissionInformation,
382    record: RecoveredRecord,
383}
384
385struct CompletedObject {
386    kind: RecordKind,
387    object_id: u64,
388    oti: ObjectTransmissionInformation,
389}
390
391fn validate_sender_config(config: FiniteObjectSenderConfig) -> Result<()> {
392    if config.symbol_size == 0 {
393        return Err(Error::InvalidConfig("RaptorQ symbol size must be nonzero"));
394    }
395    if config.max_object_bytes == 0 || config.max_object_bytes > RFC6330_MAX_TRANSFER_LENGTH {
396        return Err(Error::InvalidConfig("invalid RaptorQ maximum object size"));
397    }
398    Ok(())
399}
400
401pub(crate) fn validate_limits(limits: FiniteObjectLimits) -> Result<()> {
402    if limits.max_object_bytes == 0 || limits.max_object_bytes > RFC6330_MAX_TRANSFER_LENGTH {
403        return Err(Error::InvalidConfig("invalid RaptorQ maximum object size"));
404    }
405    if limits.max_symbol_size == 0 {
406        return Err(Error::InvalidConfig(
407            "RaptorQ maximum symbol size must be nonzero",
408        ));
409    }
410    if limits.max_concurrent_objects == 0 {
411        return Err(Error::InvalidConfig(
412            "finite-object receiver capacity must be nonzero",
413        ));
414    }
415    Ok(())
416}
417
418fn validate_oti(
419    encoded: [u8; 12],
420    limits: FiniteObjectLimits,
421) -> Result<ObjectTransmissionInformation> {
422    if encoded[5] != 0 {
423        return Err(Error::InvalidFecMetadata(
424            "RaptorQ OTI reserved byte is nonzero",
425        ));
426    }
427    let oti = ObjectTransmissionInformation::deserialize(&encoded);
428    let transfer_length = oti.transfer_length();
429    let symbol_size = oti.symbol_size();
430    let source_blocks = oti.source_blocks();
431    let sub_blocks = oti.sub_blocks();
432    let alignment = oti.symbol_alignment();
433    if transfer_length == 0 || transfer_length > limits.max_object_bytes {
434        return Err(Error::ObjectTooLarge {
435            actual: transfer_length,
436            maximum: limits.max_object_bytes,
437        });
438    }
439    if symbol_size == 0 || symbol_size > limits.max_symbol_size {
440        return Err(Error::InvalidFecMetadata(
441            "RaptorQ symbol size exceeds receiver limit",
442        ));
443    }
444    if source_blocks == 0 || sub_blocks == 0 || alignment == 0 {
445        return Err(Error::InvalidFecMetadata(
446            "RaptorQ OTI contains a zero parameter",
447        ));
448    }
449    if !symbol_size.is_multiple_of(alignment as u16) {
450        return Err(Error::InvalidFecMetadata(
451            "RaptorQ symbol alignment is invalid",
452        ));
453    }
454    let total_source_symbols = transfer_length.div_ceil(symbol_size as u64);
455    if source_blocks as u64 > total_source_symbols {
456        return Err(Error::InvalidFecMetadata(
457            "RaptorQ has more source blocks than source symbols",
458        ));
459    }
460    if sub_blocks as u32 > u32::from(symbol_size / alignment as u16) {
461        return Err(Error::InvalidFecMetadata(
462            "RaptorQ sub-block count exceeds symbol geometry",
463        ));
464    }
465    let largest_block = total_source_symbols.div_ceil(source_blocks as u64);
466    if largest_block > RFC6330_MAX_SOURCE_SYMBOLS_PER_BLOCK {
467        return Err(Error::InvalidFecMetadata(
468            "RaptorQ source block is too large",
469        ));
470    }
471    Ok(oti)
472}
cu29_logstream/object.rs

cu29_logstream/
object.rs
