cu29_logstream/
sender.rs

1use crate::rlc::RepairSchedule;
2use crate::{
3    ContinuousEncoder, CuStreamTx, CuStreamTxError, Error, FiniteObjectEncoder,
4    FiniteObjectSenderConfig, Lane, PACKET_HEADER_LEN, RecordKind, Result, StreamIdentity,
5    decode_record, encode_copperlist_record_into, encode_keyframe_and_recovery_point,
6};
7use alloc::{boxed::Box, string::ToString, vec, vec::Vec};
8use core::{fmt::Debug, marker::PhantomData};
9use cu_fec::{DensityThreshold, EncodingSymbolId, RlcConfig};
10use cu29_runtime::{copperlist::CopperList, curuntime::KeyFrame};
11use cu29_traits::{CopperListTuple, CuError, CuResult, WriteStream};
12
13/// Maximum source-symbol storage used by generated runtime integration.
14pub const DEFAULT_MAX_SYMBOL_SIZE: usize = 1128;
15/// Maximum RLC coding-window storage used by generated runtime integration.
16pub const DEFAULT_MAX_WINDOW_SYMBOLS: usize = 64;
17
18/// Concrete sender shape emitted by generated applications.
19pub type DefaultContinuousCopperListSink<P, T> =
20    ContinuousCopperListSink<P, T, DEFAULT_MAX_SYMBOL_SIZE, DEFAULT_MAX_WINDOW_SYMBOLS>;
21
22/// Configuration for the continuous CopperList sender lane.
23#[derive(Clone, Copy, Debug, PartialEq, Eq)]
24pub struct ContinuousSenderConfig {
25    pub identity: StreamIdentity,
26    pub lane: Lane,
27    pub fec: RlcConfig,
28    pub max_record_bytes: usize,
29    pub initial_esi: EncodingSymbolId,
30    /// Emit one repair after this many source symbols.
31    pub repair_every_source_symbols: usize,
32    pub first_repair_key: u16,
33    pub repair_density: DensityThreshold,
34}
35
36/// Finite-object recovery policy and the canonical manifest record repeated at every recovery point.
37#[derive(Clone, Debug, PartialEq, Eq)]
38pub struct RecoverySenderConfig {
39    pub finite: FiniteObjectSenderConfig,
40    pub manifest_record: Vec<u8>,
41    /// CopperList interval between transmitted keyframe/recovery point groups.
42    pub recovery_interval: u32,
43}
44
45/// Complete sender configuration for continuous CopperLists and recovery points.
46#[derive(Clone, Debug, PartialEq, Eq)]
47pub struct LogStreamSenderConfig {
48    /// Local timeout and adaptation policy for the feedback worker.
49    pub feedback: Option<crate::feedback::FeedbackPolicy>,
50    pub pacing: crate::PacingConfig,
51    pub continuous: ContinuousSenderConfig,
52    pub recovery: RecoverySenderConfig,
53}
54
55impl LogStreamSenderConfig {
56    pub fn validate(&self) -> Result<()> {
57        if self.continuous.identity != self.recovery.finite.identity {
58            return Err(Error::InvalidConfig(
59                "continuous and recovery sender identities differ",
60            ));
61        }
62        if self.continuous.lane == self.recovery.finite.lane {
63            return Err(Error::InvalidConfig(
64                "continuous and finite objects require distinct lanes",
65            ));
66        }
67        let manifest = decode_record(&self.recovery.manifest_record)?;
68        if manifest.kind != RecordKind::Manifest {
69            return Err(Error::InvalidConfig(
70                "recovery configuration requires a manifest record",
71            ));
72        }
73        if self.recovery.recovery_interval == 0 {
74            return Err(Error::InvalidConfig(
75                "recovery point interval must be nonzero",
76            ));
77        }
78        Ok(())
79    }
80}
81
82/// Sender-side delivery counters. Transport backpressure is observable but never blocks.
83#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
84pub struct ContinuousSenderStats {
85    pub records_encoded: u64,
86    pub source_datagrams: u64,
87    pub repair_datagrams: u64,
88    pub datagrams_sent: u64,
89    pub datagrams_dropped: u64,
90}
91
92/// Sender counters for repeated manifests, keyframes, and recovery points.
93#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
94pub struct RecoverySenderStats {
95    pub keyframes_encoded: u64,
96    pub object_datagrams: u64,
97    pub datagrams_sent: u64,
98    pub datagrams_dropped: u64,
99}
100
101/// Immediate, unpaced sink emitting keyframes and their recovery points.
102/// Intended for codec tests and custom drivers.
103/// Generated applications use `scheduled_sinks` to share a destination budget.
104pub struct RecoveryPointSink<T: CuStreamTx> {
105    transport: T,
106    encoder: FiniteObjectEncoder,
107    manifest_record: Vec<u8>,
108    manifest_object_id: u64,
109    manifest_record_digest: [u8; 32],
110    recovery_interval: u32,
111    stats: RecoverySenderStats,
112}
113
114impl<T: CuStreamTx> Debug for RecoveryPointSink<T> {
115    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
116        formatter
117            .debug_struct("RecoveryPointSink")
118            .field("transport", &self.transport)
119            .field("config", &self.encoder.config())
120            .field("stats", &self.stats)
121            .finish_non_exhaustive()
122    }
123}
124
125impl<T: CuStreamTx> RecoveryPointSink<T> {
126    pub fn new(transport: T, config: RecoverySenderConfig) -> Result<Self> {
127        if config.recovery_interval == 0 {
128            return Err(Error::InvalidConfig(
129                "recovery point interval must be nonzero",
130            ));
131        }
132        let manifest = decode_record(&config.manifest_record)?;
133        if manifest.kind != RecordKind::Manifest {
134            return Err(Error::InvalidConfig(
135                "recovery configuration requires a manifest record",
136            ));
137        }
138        let manifest_object_id = manifest.object_id;
139        let manifest_record_digest = manifest.digest;
140        Ok(Self {
141            transport,
142            encoder: FiniteObjectEncoder::new(config.finite)?,
143            manifest_record: config.manifest_record,
144            manifest_object_id,
145            manifest_record_digest,
146            recovery_interval: config.recovery_interval,
147            stats: RecoverySenderStats::default(),
148        })
149    }
150
151    pub const fn stats(&self) -> RecoverySenderStats {
152        self.stats
153    }
154
155    pub fn transport(&self) -> &T {
156        &self.transport
157    }
158
159    pub fn transport_mut(&mut self) -> &mut T {
160        &mut self.transport
161    }
162
163    pub fn into_transport(self) -> T {
164        self.transport
165    }
166
167    fn emit_record(&mut self, record: &[u8]) -> Result<()> {
168        emit_finite_record(
169            &mut self.encoder,
170            &mut self.transport,
171            &mut self.stats,
172            record,
173        )
174    }
175}
176
177fn emit_finite_record<T: CuStreamTx>(
178    encoder: &mut FiniteObjectEncoder,
179    transport: &mut T,
180    stats: &mut RecoverySenderStats,
181    record: &[u8],
182) -> Result<()> {
183    let datagrams = encoder.push_record_with(record, |datagram| {
184        stats.object_datagrams = stats.object_datagrams.saturating_add(1);
185        match transport.try_send(datagram) {
186            Ok(()) => stats.datagrams_sent = stats.datagrams_sent.saturating_add(1),
187            Err(CuStreamTxError::WouldBlock) => {
188                stats.datagrams_dropped = stats.datagrams_dropped.saturating_add(1);
189            }
190            Err(CuStreamTxError::Failed(message)) => return Err(Error::Transport(message)),
191        }
192        Ok(())
193    })?;
194    debug_assert!(datagrams > 0);
195    Ok(())
196}
197
198impl<T: CuStreamTx> WriteStream<KeyFrame> for RecoveryPointSink<T> {
199    fn log(&mut self, keyframe: &KeyFrame) -> CuResult<()> {
200        if !keyframe
201            .culistid
202            .is_multiple_of(u64::from(self.recovery_interval))
203        {
204            return Ok(());
205        }
206        let (keyframe_record, recovery_point_record) = encode_keyframe_and_recovery_point(
207            keyframe,
208            self.manifest_object_id,
209            self.manifest_record_digest,
210        )
211        .map_err(|error| CuError::from(error.to_string()))?;
212        emit_finite_record(
213            &mut self.encoder,
214            &mut self.transport,
215            &mut self.stats,
216            &self.manifest_record,
217        )
218        .and_then(|()| self.emit_record(&keyframe_record))
219        .and_then(|()| self.emit_record(&recovery_point_record))
220        .map_err(|error| CuError::from(error.to_string()))?;
221        self.stats.keyframes_encoded = self.stats.keyframes_encoded.saturating_add(1);
222        Ok(())
223    }
224}
225
226/// Immediate, unpaced CopperList sink backed by continuous RLC.
227/// Generated applications use `scheduled_sinks` for bounded packet scheduling.
228pub struct ContinuousCopperListSink<
229    P,
230    T,
231    const MAX_SYMBOL_SIZE: usize,
232    const MAX_WINDOW_SYMBOLS: usize,
233> where
234    P: CopperListTuple,
235    T: CuStreamTx,
236{
237    transport: T,
238    encoder: Box<ContinuousEncoder<MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS>>,
239    record: Vec<u8>,
240    datagram: Vec<u8>,
241    repair_schedule: RepairSchedule,
242    stats: ContinuousSenderStats,
243    _payload: PhantomData<fn() -> P>,
244}
245
246impl<P, T, const MAX_SYMBOL_SIZE: usize, const MAX_WINDOW_SYMBOLS: usize> Debug
247    for ContinuousCopperListSink<P, T, MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS>
248where
249    P: CopperListTuple,
250    T: CuStreamTx,
251{
252    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
253        formatter
254            .debug_struct("ContinuousCopperListSink")
255            .field("transport", &self.transport)
256            .field("config", &self.encoder.config())
257            .field("stats", &self.stats)
258            .finish_non_exhaustive()
259    }
260}
261
262impl<P, T, const MAX_SYMBOL_SIZE: usize, const MAX_WINDOW_SYMBOLS: usize>
263    ContinuousCopperListSink<P, T, MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS>
264where
265    P: CopperListTuple,
266    T: CuStreamTx,
267{
268    pub fn new(transport: T, config: ContinuousSenderConfig) -> Result<Self> {
269        if config.repair_every_source_symbols == 0 {
270            return Err(Error::InvalidConfig(
271                "repair interval must contain at least one source symbol",
272            ));
273        }
274        let datagram_bytes = PACKET_HEADER_LEN
275            .checked_add(config.fec.symbol_size())
276            .ok_or(Error::InvalidConfig("datagram length overflow"))?;
277        let encoder = ContinuousEncoder::new(
278            config.identity,
279            config.lane,
280            config.fec,
281            config.max_record_bytes,
282            config.initial_esi,
283        )?;
284        Ok(Self {
285            transport,
286            encoder: Box::new(encoder),
287            record: vec![0; config.max_record_bytes],
288            datagram: vec![0; datagram_bytes],
289            repair_schedule: RepairSchedule::new(
290                config.repair_every_source_symbols,
291                config.first_repair_key,
292                config.repair_density,
293            ),
294            stats: ContinuousSenderStats::default(),
295            _payload: PhantomData,
296        })
297    }
298
299    pub const fn stats(&self) -> ContinuousSenderStats {
300        self.stats
301    }
302
303    pub fn transport(&self) -> &T {
304        &self.transport
305    }
306
307    pub fn transport_mut(&mut self) -> &mut T {
308        &mut self.transport
309    }
310
311    pub fn into_transport(self) -> T {
312        self.transport
313    }
314
315    fn emit_source_record(&mut self, record_len: usize) -> Result<()> {
316        let transport = &mut self.transport;
317        let stats = &mut self.stats;
318        let (source_datagrams, repair_datagrams) = self.encoder.push_record_with_repairs(
319            &self.record[..record_len],
320            &mut self.datagram,
321            &mut |datagram| submit_datagram(transport, stats, datagram),
322            &mut self.repair_schedule,
323        )?;
324        stats.source_datagrams = stats
325            .source_datagrams
326            .saturating_add(source_datagrams as u64);
327        stats.repair_datagrams = stats
328            .repair_datagrams
329            .saturating_add(repair_datagrams as u64);
330        Ok(())
331    }
332}
333
334impl<P, T, const MAX_SYMBOL_SIZE: usize, const MAX_WINDOW_SYMBOLS: usize> WriteStream<CopperList<P>>
335    for ContinuousCopperListSink<P, T, MAX_SYMBOL_SIZE, MAX_WINDOW_SYMBOLS>
336where
337    P: CopperListTuple + Send + Sync,
338    T: CuStreamTx,
339{
340    fn log(&mut self, copperlist: &CopperList<P>) -> CuResult<()> {
341        let record_len = encode_copperlist_record_into(copperlist, &mut self.record)
342            .map_err(|error| CuError::from(error.to_string()))?;
343        self.emit_source_record(record_len)
344            .map_err(|error| CuError::from(error.to_string()))?;
345        self.stats.records_encoded = self.stats.records_encoded.saturating_add(1);
346        Ok(())
347    }
348}
349
350fn submit_datagram<T: CuStreamTx>(
351    transport: &mut T,
352    stats: &mut ContinuousSenderStats,
353    datagram: &[u8],
354) -> Result<()> {
355    match transport.try_send(datagram) {
356        Ok(()) => stats.datagrams_sent = stats.datagrams_sent.saturating_add(1),
357        Err(CuStreamTxError::WouldBlock) => {
358            stats.datagrams_dropped = stats.datagrams_dropped.saturating_add(1);
359        }
360        Err(CuStreamTxError::Failed(message)) => return Err(Error::Transport(message)),
361    }
362    Ok(())
363}
cu29_logstream/sender.rs

cu29_logstream/
sender.rs
