cu29_logstream/
router.rs

1//! Bounded ground-side session discovery and decoder routing.
2
3use crate::{
4    ContinuousDecoder, ContinuousReceiveEvent, CuStreamRx, CuStreamRxError, Error, FecScheme,
5    FiniteObjectDecoder, FiniteObjectLimits, Lane, ReceiveError, ReceiverLimits, RecoveredRecord,
6    Result, SessionManifest, StreamIdentity, WirePacketRef,
7};
8use alloc::{collections::VecDeque, vec::Vec};
9
10#[derive(Clone, Copy, Debug, PartialEq, Eq)]
11pub struct SessionRouterLimits {
12    pub max_sessions: usize,
13    /// Per-session pre-manifest datagrams, each bounded by MAX_SYMBOL_SIZE + header.
14    pub max_startup_packets: usize,
15    /// Completed keyframe/recovery point records retained per session, in addition to FEC storage.
16    pub max_recovery_records: usize,
17    pub max_pending_events: usize,
18    pub max_record_bytes: usize,
19    pub max_buffered_records: usize,
20    pub equation_capacity: usize,
21    pub finite_objects: FiniteObjectLimits,
22}
23
24impl Default for SessionRouterLimits {
25    /// One sender using the 1200-byte-MTU, 64-symbol streaming profile.
26    fn default() -> Self {
27        Self {
28            max_sessions: 1,
29            max_startup_packets: 64,
30            max_recovery_records: 8,
31            max_pending_events: 64,
32            max_record_bytes: 4096,
33            max_buffered_records: 64,
34            equation_capacity: 64,
35            finite_objects: FiniteObjectLimits::new(65536, 1128, 4),
36        }
37    }
38}
39
40#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
41pub struct SessionRouterStats {
42    pub datagrams_seen: usize,
43    pub malformed_datagrams: usize,
44    pub datagrams_before_manifest: usize,
45    pub sessions_discovered: usize,
46    pub manifests_accepted: usize,
47    pub startup_packets_dropped: usize,
48}
49
50/// A decoded manifest bound to its verified, immutable wire record.
51#[derive(Clone, Debug, PartialEq, Eq)]
52pub struct ReceivedManifest {
53    manifest: SessionManifest,
54    record: RecoveredRecord,
55}
56
57impl ReceivedManifest {
58    /// Takes ownership of canonical bytes and verifies their manifest envelope.
59    pub fn decode_record(bytes: Vec<u8>) -> Result<Self> {
60        Self::from_record(RecoveredRecord::from_bytes(bytes)?)
61    }
62
63    fn from_record(record: RecoveredRecord) -> Result<Self> {
64        let manifest = SessionManifest::from_decoded(record.decoded())?;
65        Ok(Self { manifest, record })
66    }
67
68    pub fn manifest(&self) -> &SessionManifest {
69        &self.manifest
70    }
71
72    pub fn record(&self) -> &RecoveredRecord {
73        &self.record
74    }
75}
76
77/// Session events borrow receiver storage during delivery. Use `to_owned` only
78/// when an application deliberately needs to retain encoded history.
79#[derive(Clone, Debug, PartialEq, Eq)]
80pub enum SessionEvent<R = RecoveredRecord, M = ReceivedManifest> {
81    Manifest(M),
82    /// Emitted only after the recovery point, manifest and keyframe agree by digest and id.
83    VerifiedRecoveryPoint {
84        identity: StreamIdentity,
85        recovery_point: crate::RecoveryPoint,
86        recovery_record: R,
87        keyframe: R,
88    },
89    ContinuousRecord {
90        identity: StreamIdentity,
91        record: R,
92    },
93    Gap {
94        identity: StreamIdentity,
95        gap: crate::CopperListGap,
96    },
97    Object {
98        identity: StreamIdentity,
99        record: R,
100    },
101}
102/// A callback-scoped view; retaining it requires an explicit allocating `to_owned` call.
103pub type SessionEventRef<'a> = SessionEvent<&'a RecoveredRecord, &'a ReceivedManifest>;
104impl SessionEventRef<'_> {
105    pub fn to_owned(&self) -> SessionEvent {
106        match self {
107            Self::Manifest(manifest) => SessionEvent::Manifest((*manifest).clone()),
108            Self::VerifiedRecoveryPoint {
109                identity,
110                recovery_point,
111                recovery_record,
112                keyframe,
113            } => SessionEvent::VerifiedRecoveryPoint {
114                identity: *identity,
115                recovery_point: recovery_point.clone(),
116                recovery_record: (*recovery_record).clone(),
117                keyframe: (*keyframe).clone(),
118            },
119            Self::ContinuousRecord { identity, record } => SessionEvent::ContinuousRecord {
120                identity: *identity,
121                record: (*record).clone(),
122            },
123            Self::Gap { identity, gap } => SessionEvent::Gap {
124                identity: *identity,
125                gap: *gap,
126            },
127            Self::Object { identity, record } => SessionEvent::Object {
128                identity: *identity,
129                record: (*record).clone(),
130            },
131        }
132    }
133}
134
135/// Discovers sender identities and routes packets using the recovered manifest.
136/// Event payloads remain in receiver storage until the callback succeeds.
137pub struct SessionRouter<
138    const MAX_SYMBOL_SIZE: usize,
139    const MAX_WINDOW_SYMBOLS: usize,
140    const MAX_EQUATIONS: usize,
141> {
142    limits: SessionRouterLimits,
143    sessions: Vec<RoutedSession<MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS, MAX_EQUATIONS>>,
144    pending: VecDeque<PendingEvent>,
145    stats: SessionRouterStats,
146}
147
148struct RecoveryRecord {
149    record: RecoveredRecord,
150    point: Option<crate::RecoveryPoint>,
151}
152// Cache indices remain stable: pending delivery completes before another control
153// packet can mutate a session'session cache. A finite packet completes at most one object.
154enum PendingEvent {
155    Manifest(usize),
156    CachedObject {
157        session: usize,
158        slot: usize,
159    },
160    Object {
161        identity: StreamIdentity,
162        record: RecoveredRecord,
163    },
164    RecoveryPoint {
165        session: usize,
166        point: usize,
167        keyframe: usize,
168    },
169    Gap {
170        identity: StreamIdentity,
171        gap: crate::CopperListGap,
172    },
173}
174
175struct RoutedSession<const S: usize, const W: usize, const E: usize> {
176    identity: StreamIdentity,
177    finite: FiniteObjectDecoder,
178    structured: FiniteObjectDecoder,
179    structured_seen: Option<(u64, u64)>,
180    manifest: Option<ReceivedManifest>,
181    startup: VecDeque<Vec<u8>>,
182    recovery: Vec<RecoveryRecord>,
183    recovery_dirty: bool,
184    last_recovery_point: Option<u64>,
185    delivered_record: bool,
186    received_bytes: u64,
187    received_packets: u64,
188    continuous: Option<ContinuousDecoder<S, W, E>>,
189}
190
191impl<const MAX_SYMBOL_SIZE: usize, const MAX_WINDOW_SYMBOLS: usize, const MAX_EQUATIONS: usize>
192    SessionRouter<MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS, MAX_EQUATIONS>
193{
194    pub fn new(limits: SessionRouterLimits) -> Result<Self> {
195        if limits.max_sessions == 0
196            || limits.max_pending_events < 3
197            || limits.max_recovery_records < 2
198            || limits.max_record_bytes == 0
199            || limits.max_buffered_records == 0
200            || limits.equation_capacity == 0
201        {
202            return Err(Error::InvalidConfig("invalid session router capacities"));
203        }
204        Ok(Self {
205            limits,
206            sessions: Vec::with_capacity(limits.max_sessions),
207            pending: VecDeque::with_capacity(limits.max_pending_events),
208            stats: SessionRouterStats::default(),
209        })
210    }
211
212    pub const fn stats(&self) -> SessionRouterStats {
213        self.stats
214    }
215
216    /// Receiver-worker snapshot. Malformed datagrams are router-wide because their identity cannot be trusted.
217    pub fn feedback_counters(
218        &self,
219        identity: StreamIdentity,
220    ) -> Option<crate::feedback::ReceiverReport> {
221        let session = self.sessions.iter().find(|s| s.identity == identity)?;
222        let decoder = session.continuous.as_ref()?;
223        let stats = decoder.stats();
224        let occupancy = decoder.occupancy();
225        Some(crate::feedback::ReceiverReport {
226            received_bytes: session.received_bytes,
227            received_packets: session.received_packets,
228            sources: decoder.source_outcomes(),
229            invalid_packets: self.stats.malformed_datagrams as u64
230                + stats.invalid_datagrams as u64
231                + stats.inconsistent_datagrams as u64,
232            duplicate_packets: stats.duplicate_symbols as u64,
233            expired_records: stats.records_expired as u64,
234            buffered_records: (occupancy.incomplete_records + occupancy.ready_records)
235                .min(u32::MAX as usize) as u32,
236            record_capacity: self.limits.max_buffered_records.min(u32::MAX as usize) as u32,
237            request_recovery: session.last_recovery_point.is_none(),
238            ..Default::default()
239        })
240    }
241
242    pub fn session_manifest(&self, identity: StreamIdentity) -> Option<&SessionManifest> {
243        self.sessions
244            .iter()
245            .find(|session| session.identity == identity)
246            .and_then(|session| session.manifest.as_ref())
247            .map(ReceivedManifest::manifest)
248    }
249
250    pub fn try_receive<T: CuStreamRx, E>(
251        &mut self,
252        transport: &mut T,
253        packet: &mut [u8],
254        emit: impl FnMut(SessionEventRef<'_>) -> core::result::Result<(), E>,
255    ) -> core::result::Result<bool, ReceiveError<E>> {
256        let len = match transport.try_recv(packet) {
257            Ok(Some(len)) => len,
258            Ok(None) => return Ok(false),
259            Err(CuStreamRxError::BufferTooSmall { needed }) => {
260                return Err(Error::BufferTooSmall {
261                    needed,
262                    available: packet.len(),
263                }
264                .into());
265            }
266            Err(CuStreamRxError::Failed(message)) => return Err(Error::Transport(message).into()),
267        };
268        if len > packet.len() {
269            return Err(Error::BufferTooSmall {
270                needed: len,
271                available: packet.len(),
272            }
273            .into());
274        }
275        self.receive_datagram(&packet[..len], emit)?;
276        Ok(true)
277    }
278
279    /// Accepts one complete packet whose integrity was verified by the carrier.
280    /// Direct callers must uphold the same contract as [`CuStreamRx`].
281    pub fn receive_datagram<E>(
282        &mut self,
283        datagram: &[u8],
284        mut emit: impl FnMut(SessionEventRef<'_>) -> core::result::Result<(), E>,
285    ) -> core::result::Result<(), ReceiveError<E>> {
286        self.drain_events(&mut emit)?;
287        self.stats.datagrams_seen += 1;
288        let packet = match WirePacketRef::decode(datagram) {
289            Ok(p) => p,
290            Err(_) => {
291                self.stats.malformed_datagrams += 1;
292                return Ok(());
293            }
294        };
295        if (packet.header.lane == Lane::StructuredLog
296            || packet.header.record_kind == crate::RecordKind::StructuredLog)
297            && !(packet.header.lane == Lane::StructuredLog
298                && packet.header.record_kind == crate::RecordKind::StructuredLog
299                && packet.header.fec_scheme == FecScheme::RaptorQ)
300        {
301            self.stats.malformed_datagrams += 1;
302            return Ok(());
303        }
304        let identity = StreamIdentity {
305            session_id: packet.header.session_id,
306            sender_id: packet.header.sender_id,
307        };
308        let index = match self
309            .sessions
310            .iter()
311            .position(|session| session.identity == identity)
312        {
313            Some(i) => i,
314            None if (packet.header.fec_scheme == FecScheme::RaptorQ
315                && packet.header.lane == Lane::Control)
316                || (packet.header.lane == Lane::StructuredLog
317                    && packet.header.record_kind == crate::RecordKind::StructuredLog
318                    && packet.header.fec_scheme == FecScheme::RaptorQ)
319                || (packet.header.lane == Lane::ReplayCritical
320                    && packet.header.record_kind == crate::RecordKind::CopperList) =>
321            {
322                if self.sessions.len() == self.limits.max_sessions {
323                    return Err(Error::TooManySessions {
324                        maximum: self.limits.max_sessions,
325                    }
326                    .into());
327                }
328                self.sessions.push(RoutedSession {
329                    identity,
330                    finite: FiniteObjectDecoder::new(
331                        identity,
332                        Lane::Control,
333                        self.limits.finite_objects,
334                    )?,
335                    structured: FiniteObjectDecoder::new(
336                        identity,
337                        Lane::StructuredLog,
338                        self.limits.finite_objects,
339                    )?,
340                    structured_seen: None,
341                    manifest: None,
342                    startup: VecDeque::with_capacity(self.limits.max_startup_packets),
343                    recovery: Vec::with_capacity(self.limits.max_recovery_records),
344                    recovery_dirty: false,
345                    last_recovery_point: None,
346                    delivered_record: false,
347                    received_bytes: 0,
348                    received_packets: 0,
349                    continuous: None,
350                });
351                self.stats.sessions_discovered += 1;
352                self.sessions.len() - 1
353            }
354            None => {
355                self.stats.datagrams_before_manifest += 1;
356                return Ok(());
357            }
358        };
359        self.sessions[index].received_bytes += datagram.len() as u64;
360        self.sessions[index].received_packets += 1;
361        if packet.header.lane == Lane::StructuredLog {
362            if self.sessions[index].manifest.is_some() {
363                self.sessions[index].structured.receive_packet(packet)?;
364                if let Some(record) = self.sessions[index].structured.pop_record() {
365                    self.receive_object(index, record)?;
366                }
367            } else {
368                self.buffer_startup(index, datagram);
369            }
370        } else if packet.header.fec_scheme == FecScheme::RaptorQ {
371            self.sessions[index].finite.receive_packet(packet)?;
372            if let Some(record) = self.sessions[index].finite.pop_record() {
373                self.receive_object(index, record)?;
374            }
375        } else if self.sessions[index].continuous.is_some() {
376            Self::receive_continuous(&mut self.sessions[index], packet, &mut emit)?;
377        } else {
378            self.buffer_startup(index, datagram);
379        }
380        self.drain_events(&mut emit)
381    }
382    /// Retries pending delivery without cloning encoded records. A consumer error
383    /// retains the same bytes until a later call succeeds.
384    fn buffer_startup(&mut self, index: usize, datagram: &[u8]) {
385        self.stats.datagrams_before_manifest += 1;
386        let startup = &mut self.sessions[index].startup;
387        if datagram.len() <= MAX_SYMBOL_SIZE + crate::PACKET_HEADER_LEN
388            && startup.len() < self.limits.max_startup_packets
389        {
390            startup.push_back(datagram.to_vec());
391        } else {
392            self.stats.startup_packets_dropped += 1;
393        }
394    }
395
396    pub fn drain_events<E>(
397        &mut self,
398        emit: &mut impl FnMut(SessionEventRef<'_>) -> core::result::Result<(), E>,
399    ) -> core::result::Result<(), ReceiveError<E>> {
400        loop {
401            while let Some(pending) = self.pending.front() {
402                let event = match pending {
403                    PendingEvent::Manifest(i) => {
404                        SessionEvent::Manifest(self.sessions[*i].manifest.as_ref().unwrap())
405                    }
406                    PendingEvent::CachedObject { session, slot } => SessionEvent::Object {
407                        identity: self.sessions[*session].identity,
408                        record: &self.sessions[*session].recovery[*slot].record,
409                    },
410                    PendingEvent::Object { identity, record } => SessionEvent::Object {
411                        identity: *identity,
412                        record,
413                    },
414                    PendingEvent::Gap { identity, gap } => SessionEvent::Gap {
415                        identity: *identity,
416                        gap: *gap,
417                    },
418                    PendingEvent::RecoveryPoint {
419                        session,
420                        point,
421                        keyframe,
422                    } => {
423                        let session = &self.sessions[*session];
424                        SessionEvent::VerifiedRecoveryPoint {
425                            identity: session.identity,
426                            recovery_point: session.recovery[*point]
427                                .point
428                                .as_ref()
429                                .unwrap()
430                                .clone(),
431                            recovery_record: &session.recovery[*point].record,
432                            keyframe: &session.recovery[*keyframe].record,
433                        }
434                    }
435                };
436                emit(event).map_err(ReceiveError::Consumer)?;
437                self.pending.pop_front();
438            }
439            for session in &mut self.sessions {
440                if let Some(decoder) = &mut session.continuous {
441                    decoder.drain_events(|e| {
442                        Self::deliver_continuous(
443                            session.identity,
444                            session.last_recovery_point,
445                            &mut session.delivered_record,
446                            e,
447                            emit,
448                        )
449                    })?;
450                }
451            }
452            if let Some(index) = self
453                .sessions
454                .iter()
455                .position(|session| session.continuous.is_some() && !session.startup.is_empty())
456            {
457                let packet = self.sessions[index].startup.pop_front().unwrap();
458                let decoded = WirePacketRef::decode(&packet)?;
459                if decoded.header.lane == Lane::StructuredLog {
460                    self.sessions[index].structured.receive_packet(decoded)?;
461                    if let Some(record) = self.sessions[index].structured.pop_record() {
462                        self.receive_object(index, record)?;
463                    }
464                } else {
465                    Self::receive_continuous(&mut self.sessions[index], decoded, emit)?;
466                }
467                continue;
468            }
469            for index in 0..self.sessions.len() {
470                if self.sessions[index].recovery_dirty {
471                    self.verify_recovery(index)?;
472                }
473                if !self.pending.is_empty() {
474                    break;
475                }
476            }
477            if self.pending.is_empty() {
478                return Ok(());
479            }
480        }
481    }
482
483    fn deliver_continuous<E>(
484        identity: StreamIdentity,
485        last_recovery_point: Option<u64>,
486        delivered: &mut bool,
487        event: ContinuousReceiveEvent<'_>,
488        emit: &mut impl FnMut(SessionEventRef<'_>) -> core::result::Result<(), E>,
489    ) -> core::result::Result<(), E> {
490        match event {
491            ContinuousReceiveEvent::Record(record) => {
492                emit(SessionEvent::ContinuousRecord { identity, record })?;
493                *delivered = true;
494            }
495            ContinuousReceiveEvent::Gap(mut gap) => {
496                if !*delivered && last_recovery_point.is_none() {
497                    gap.reason = crate::GapReason::LateJoin;
498                }
499                emit(SessionEvent::Gap { identity, gap })?;
500            }
501        }
502        Ok(())
503    }
504
505    fn receive_continuous<E>(
506        session: &mut RoutedSession<MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS, MAX_EQUATIONS>,
507        packet: WirePacketRef<'_>,
508        emit: &mut impl FnMut(SessionEventRef<'_>) -> core::result::Result<(), E>,
509    ) -> core::result::Result<(), ReceiveError<E>> {
510        session
511            .continuous
512            .as_mut()
513            .unwrap()
514            .receive_packet(packet, |e| {
515                Self::deliver_continuous(
516                    session.identity,
517                    session.last_recovery_point,
518                    &mut session.delivered_record,
519                    e,
520                    emit,
521                )
522            })
523    }
524
525    fn receive_object(&mut self, index: usize, mut record: RecoveredRecord) -> Result<()> {
526        if record.decoded().kind == crate::RecordKind::StructuredLog {
527            // Accept reordered records inside a fixed 64-record window and suppress
528            // late duplicates even after the finite decoder evicts its object cache.
529            let id = record.decoded().object_id;
530            let seen = self.sessions[index].structured_seen;
531            let updated = match seen {
532                None => (id, 1),
533                Some((latest, bits)) if id > latest => (
534                    id,
535                    bits.checked_shl((id - latest).min(64) as u32).unwrap_or(0) | 1,
536                ),
537                Some((latest, bits)) => {
538                    let age = latest - id;
539                    if age >= 64 || bits & (1 << age) != 0 {
540                        return Ok(());
541                    }
542                    (latest, bits | (1 << age))
543                }
544            };
545            self.push_event(PendingEvent::Object {
546                identity: self.sessions[index].identity,
547                record,
548            })?;
549            self.sessions[index].structured_seen = Some(updated);
550            return Ok(());
551        }
552        if record.decoded().kind == crate::RecordKind::Manifest {
553            if let Some(existing) = &self.sessions[index].manifest {
554                if existing.record.decoded().digest != record.decoded().digest {
555                    return Err(Error::InconsistentObject);
556                }
557                return Ok(());
558            }
559            let manifest = ReceivedManifest::from_record(record)?;
560            if manifest.manifest.identity != self.sessions[index].identity {
561                return Err(Error::InconsistentObject);
562            }
563            self.sessions[index].continuous = Some(Self::decoder_from_manifest(
564                self.limits,
565                &manifest.manifest,
566            )?);
567            self.sessions[index].manifest = Some(manifest);
568            self.sessions[index].recovery_dirty = true;
569            self.stats.manifests_accepted += 1;
570            self.push_event(PendingEvent::Manifest(index))?;
571        } else {
572            let point = match record.decoded().kind {
573                crate::RecordKind::KeyFrame => {
574                    record.validate_keyframe()?;
575                    None
576                }
577                crate::RecordKind::RecoveryPoint => {
578                    Some(crate::decode_recovery_point(record.decoded().payload)?)
579                }
580                _ => {
581                    return self.push_event(PendingEvent::Object {
582                        identity: self.sessions[index].identity,
583                        record,
584                    });
585                }
586            };
587            let session = &mut self.sessions[index];
588            if session.recovery.len() == self.limits.max_recovery_records {
589                session.recovery.remove(0);
590            }
591            let slot = session.recovery.len();
592            session.recovery.push(RecoveryRecord { record, point });
593            session.recovery_dirty = true;
594            self.push_event(PendingEvent::CachedObject {
595                session: index,
596                slot,
597            })?;
598        }
599        Ok(())
600    }
601
602    fn verify_recovery(&mut self, index: usize) -> Result<()> {
603        let session = &mut self.sessions[index];
604        let Some(manifest) = &session.manifest else {
605            return Ok(());
606        };
607        let mut candidate = None;
608        for (point_index, entry) in session.recovery.iter().enumerate() {
609            let Some(point) = &entry.point else {
610                continue;
611            };
612            if point.copperlist_id != entry.record.decoded().object_id
613                || !point.references_manifest(manifest.record.decoded())
614                || session
615                    .last_recovery_point
616                    .is_some_and(|id| point.copperlist_id <= id)
617            {
618                continue;
619            }
620            for (kf_index, kf) in session.recovery.iter().enumerate() {
621                if point.references_keyframe(kf.record.decoded())
622                    && kf.record.keyframe_id() == Some(point.copperlist_id)
623                    && candidate.is_none_or(|(_, _, id)| point.copperlist_id > id)
624                {
625                    candidate = Some((point_index, kf_index, point.copperlist_id));
626                }
627            }
628        }
629        session.recovery_dirty = false;
630        let Some((point, keyframe, id)) = candidate else {
631            return Ok(());
632        };
633        let decoder = session.continuous.as_mut().unwrap();
634        let next = decoder.next_object_id();
635        let needed = 1 + usize::from(id > next);
636        if self.pending.len() + needed > self.limits.max_pending_events {
637            session.recovery_dirty = true;
638            return Err(Error::TooManyRecords {
639                actual: self.pending.len() + needed,
640                maximum: self.limits.max_pending_events,
641            });
642        }
643        if id > next {
644            self.pending.push_back(PendingEvent::Gap {
645                identity: session.identity,
646                gap: crate::CopperListGap {
647                    first_id: next,
648                    last_id: id - 1,
649                    reason: if session.delivered_record {
650                        crate::GapReason::RecoveryPoint
651                    } else {
652                        crate::GapReason::LateJoin
653                    },
654                },
655            });
656            decoder.resume_at(id);
657        }
658        session.last_recovery_point = Some(id);
659        self.pending.push_back(PendingEvent::RecoveryPoint {
660            session: index,
661            point,
662            keyframe,
663        });
664        Ok(())
665    }
666
667    pub fn finish_through<E>(
668        &mut self,
669        identity: StreamIdentity,
670        last_id: u64,
671        mut emit: impl FnMut(SessionEventRef<'_>) -> core::result::Result<(), E>,
672    ) -> core::result::Result<(), ReceiveError<E>> {
673        self.drain_events(&mut emit)?;
674        let session = self
675            .sessions
676            .iter_mut()
677            .find(|session| session.identity == identity)
678            .ok_or(Error::InvalidConfig("unknown session"))?;
679        session
680            .continuous
681            .as_mut()
682            .ok_or(Error::InvalidConfig("session has no manifest"))?
683            .finish_through(last_id, |event| match event {
684                ContinuousReceiveEvent::Record(record) => {
685                    emit(SessionEvent::ContinuousRecord { identity, record })
686                }
687                ContinuousReceiveEvent::Gap(gap) => emit(SessionEvent::Gap { identity, gap }),
688            })
689    }
690
691    fn decoder_from_manifest(
692        limits: SessionRouterLimits,
693        manifest: &SessionManifest,
694    ) -> Result<ContinuousDecoder<MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS, MAX_EQUATIONS>> {
695        if manifest.requirements.max_record_bytes > limits.max_record_bytes as u64 {
696            return Err(Error::ObjectTooLarge {
697                actual: manifest.requirements.max_record_bytes,
698                maximum: limits.max_record_bytes as u64,
699            });
700        }
701        if usize::from(manifest.requirements.symbol_size) > MAX_SYMBOL_SIZE
702            || usize::from(manifest.requirements.window_symbols) > MAX_WINDOW_SYMBOLS
703            || usize::from(manifest.requirements.window_symbols) > limits.max_buffered_records
704        {
705            return Err(Error::InvalidConfig(
706                "session manifest exceeds receiver-local limits",
707            ));
708        }
709        let mut decoder = ContinuousDecoder::new(
710            manifest.identity,
711            Lane::ReplayCritical,
712            manifest.requirements.rlc_config()?,
713            limits.equation_capacity,
714            0,
715            ReceiverLimits::new(
716                manifest.requirements.max_record_bytes as usize,
717                limits.max_buffered_records,
718            ),
719        )?;
720        if manifest.requirements.feedback.is_some() {
721            decoder.enable_feedback();
722        }
723        Ok(decoder)
724    }
725
726    fn push_event(&mut self, event: PendingEvent) -> Result<()> {
727        if self.pending.len() == self.limits.max_pending_events {
728            return Err(Error::TooManyRecords {
729                actual: self.pending.len() + 1,
730                maximum: self.limits.max_pending_events,
731            });
732        }
733        self.pending.push_back(event);
734        Ok(())
735    }
736}
cu29_logstream/router.rs

cu29_logstream/
router.rs
