cu29_logstream/
manifest.rs

1//! Local sender plans and self-describing receiver requirements.
2
3use crate::{
4    ContinuousSenderConfig, DensityThreshold, EncodingSymbolId, Error, Field,
5    FiniteObjectSenderConfig, Lane, LogStreamSenderConfig, PACKET_HEADER_LEN, RecordKind,
6    RecoverySenderConfig, Result, RlcConfig, StreamIdentity,
7    decode_record as decode_semantic_record, encode_record,
8};
9use alloc::{
10    string::{String, ToString},
11    vec::Vec,
12};
13use bincode::{Decode, Encode};
14use cu29_runtime::config::{LogStreamDestinationConfig, LogStreamRepairDensity, LogStreamRlcField};
15use cu29_traits::TaskOutputSpec;
16use cu29_unifiedlog::SectionContext;
17
18/// Link and codec policy after RON validation and MTU resolution.
19#[derive(Clone, Debug, PartialEq, Eq, Encode, Decode)]
20pub struct LogStreamPlan {
21    pub feedback: Option<crate::feedback::FeedbackPolicy>,
22    pub destination_id: String,
23    pub mtu_bytes: u16,
24    pub symbol_size: u16,
25    pub bitrate_bps: u64,
26    pub memory_budget_kib: u32,
27    pub max_latency_ms: u32,
28    pub burst_packets: u32,
29    pub continuous: ResolvedContinuousFec,
30    pub objects: ResolvedObjectFec,
31    /// Recovery interval in CopperLists.
32    pub recovery_interval: u32,
33    pub max_record_bytes: u64,
34}
35
36#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
37pub struct ResolvedContinuousFec {
38    pub field: ResolvedRlcField,
39    pub window_symbols: u16,
40    pub repair_every_source_symbols: u16,
41    pub repair_density: u8,
42}
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
45pub enum ResolvedRlcField {
46    Gf2,
47    Gf256,
48}
49
50#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
51pub struct ResolvedObjectFec {
52    pub max_object_bytes: u64,
53    pub repair_symbols_per_block: u32,
54}
55
56/// Minimal application schema required to interpret recovered CopperList slots.
57#[derive(Clone, Debug, PartialEq, Eq, Encode, Decode)]
58pub struct ApplicationSchema {
59    pub outputs: Vec<ApplicationOutputSchema>,
60    /// Per-output reconstruction flags (`true` means reconstruct); empty means full capture.
61    pub reconstruction: Vec<bool>,
62}
63
64#[derive(Clone, Debug, PartialEq, Eq, Encode, Decode)]
65pub struct ApplicationOutputSchema {
66    pub task_id: String,
67    pub message_type: String,
68    pub payload_type: String,
69}
70
71impl ApplicationSchema {
72    pub fn from_output_specs(specs: &[TaskOutputSpec]) -> Self {
73        Self {
74            reconstruction: Vec::new(),
75            outputs: specs
76                .iter()
77                .map(|spec| ApplicationOutputSchema {
78                    task_id: spec.task_id.into(),
79                    message_type: spec.msg_type.into(),
80                    payload_type: spec.payload_type_path().into(),
81                })
82                .collect(),
83        }
84    }
85}
86
87/// Continuous decoder geometry and record bounds carried by a session manifest.
88/// Receiver-local limits still cap admission; sender pacing and repair policy stay local.
89#[derive(Clone, Copy, Debug, PartialEq, Eq, Encode, Decode)]
90pub struct ReceiverRequirements {
91    /// Present when the sender accepts advisory return reports.
92    pub feedback: Option<crate::feedback::FeedbackRequirements>,
93    /// Bytes in each continuous FEC symbol, including its protected fragment header.
94    pub symbol_size: u16,
95    /// Finite field used by the continuous RLC lane.
96    pub field: ResolvedRlcField,
97    /// Maximum symbols retained in the continuous coding window.
98    pub window_symbols: u16,
99    /// Maximum complete framed CopperList record size.
100    pub max_record_bytes: u64,
101}
102
103impl ReceiverRequirements {
104    /// Validates untrusted geometry before constructing a continuous decoder.
105    pub fn validate(&self) -> Result<()> {
106        if self
107            .feedback
108            .is_some_and(|feedback| feedback.report_interval_ms == 0)
109        {
110            return Err(Error::InvalidConfig(
111                "feedback report interval must be nonzero",
112            ));
113        }
114        if usize::from(self.symbol_size) <= crate::rlc::FRAGMENT_HEADER_LEN {
115            return Err(Error::InvalidConfig(
116                "symbol leaves no room for continuous record bytes",
117            ));
118        }
119        if usize::from(self.symbol_size) > crate::DEFAULT_MAX_SYMBOL_SIZE
120            || usize::from(self.window_symbols) > crate::DEFAULT_MAX_WINDOW_SYMBOLS
121        {
122            return Err(Error::InvalidConfig(
123                "receiver requirements exceed supported FEC capacities",
124            ));
125        }
126        if self.window_symbols == 0 || self.max_record_bytes == 0 {
127            return Err(Error::InvalidConfig(
128                "receiver requirements contain a zero bound",
129            ));
130        }
131        if self.max_record_bytes > u64::from(u32::MAX) {
132            return Err(Error::InvalidConfig("maximum record size exceeds u32"));
133        }
134        usize::try_from(self.max_record_bytes)
135            .map_err(|_| Error::InvalidConfig("record bound exceeds usize"))?;
136        self.rlc_config()?;
137        Ok(())
138    }
139
140    /// Constructs the FEC configuration used by the continuous decoder.
141    pub fn rlc_config(&self) -> Result<RlcConfig> {
142        RlcConfig::new(
143            usize::from(self.symbol_size),
144            usize::from(self.window_symbols),
145            self.field.into(),
146        )
147        .map_err(Into::into)
148    }
149}
150
151/// Control object carrying receiver requirements and application schema checks.
152///
153/// No content or protocol version is encoded. Receivers must use the matching
154/// application's decoder; these schema checks do not establish binary compatibility.
155#[derive(Clone, Debug, PartialEq, Eq, Encode, Decode)]
156pub struct SessionManifest {
157    pub identity: StreamIdentity,
158    pub requirements: ReceiverRequirements,
159    pub application_schema: ApplicationSchema,
160    /// Onboard section identity for this construction; repeated for late joins.
161    pub context: SectionContext,
162}
163
164impl SessionManifest {
165    pub fn new(
166        identity: StreamIdentity,
167        requirements: ReceiverRequirements,
168        application_schema: ApplicationSchema,
169        context: SectionContext,
170    ) -> Self {
171        Self {
172            identity,
173            requirements,
174            application_schema,
175            context,
176        }
177    }
178
179    pub fn encode_record(&self) -> Result<Vec<u8>> {
180        let payload = bincode::encode_to_vec(self, bincode::config::standard())
181            .map_err(|error| Error::Codec(error.to_string()))?;
182        encode_record(RecordKind::Manifest, 0, &payload)
183    }
184
185    pub fn decode_record(record: &[u8]) -> Result<Self> {
186        let record = decode_semantic_record(record)?;
187        Self::from_decoded(record)
188    }
189
190    pub(crate) fn from_decoded(record: crate::DecodedRecord<'_>) -> Result<Self> {
191        if record.kind != RecordKind::Manifest {
192            return Err(Error::InvalidConfig("record is not a session manifest"));
193        }
194        let (manifest, consumed): (Self, usize) =
195            bincode::decode_from_slice(record.payload, bincode::config::standard())
196                .map_err(|error| Error::Codec(error.to_string()))?;
197        if consumed != record.payload.len() {
198            return Err(Error::Codec("session manifest has trailing bytes".into()));
199        }
200        manifest.requirements.validate()?;
201        if manifest.context.instance_id != manifest.identity.sender_id {
202            return Err(Error::InvalidConfig(
203                "manifest instance differs from sender identity",
204            ));
205        }
206        Ok(manifest)
207    }
208}
209
210impl LogStreamPlan {
211    /// Resolves one RON destination into local sender policy and wire geometry.
212    /// Symbol size is bounded by both the link MTU and preallocated sender storage;
213    /// a larger MTU does not increase the runtime's memory footprint.
214    pub fn resolve(config: &LogStreamDestinationConfig) -> Result<Self> {
215        let symbol_size = config
216            .link
217            .mtu_bytes
218            .checked_sub(PACKET_HEADER_LEN as u16)
219            .ok_or(Error::InvalidConfig("MTU does not fit the packet header"))?
220            .min(crate::DEFAULT_MAX_SYMBOL_SIZE as u16);
221        if usize::from(config.fec.continuous.window_symbols) > crate::DEFAULT_MAX_WINDOW_SYMBOLS {
222            return Err(Error::InvalidConfig(
223                "RLC window exceeds the generated sender capacity",
224            ));
225        }
226        let field = match config.fec.continuous.field {
227            LogStreamRlcField::Gf2 => ResolvedRlcField::Gf2,
228            LogStreamRlcField::Gf256 => ResolvedRlcField::Gf256,
229        };
230        let repair_density = match config.fec.continuous.repair_density {
231            LogStreamRepairDensity::Full => DensityThreshold::FULL.get(),
232            LogStreamRepairDensity::Threshold(value) => value,
233        };
234        DensityThreshold::new(repair_density)?;
235        let plan = Self {
236            feedback: config
237                .feedback
238                .as_ref()
239                .map(crate::feedback::FeedbackPolicy::from),
240            destination_id: config.id.clone(),
241            mtu_bytes: config.link.mtu_bytes,
242            symbol_size,
243            bitrate_bps: config.link.bitrate_bps,
244            memory_budget_kib: config.link.memory_budget_kib,
245            max_latency_ms: config.link.max_latency_ms,
246            burst_packets: config.link.burst_packets,
247            continuous: ResolvedContinuousFec {
248                field,
249                window_symbols: config.fec.continuous.window_symbols,
250                repair_every_source_symbols: config.fec.continuous.repair_every_source_symbols,
251                repair_density,
252            },
253            objects: ResolvedObjectFec {
254                max_object_bytes: config.fec.objects.max_object_bytes,
255                repair_symbols_per_block: config.fec.objects.repair_symbols_per_block,
256            },
257            recovery_interval: config.recovery_interval,
258            max_record_bytes: config.max_record_bytes,
259        };
260        plan.validate()?;
261        Ok(plan)
262    }
263
264    /// Validates local sender policy, capacities, and memory budget.
265    pub fn validate(&self) -> Result<()> {
266        if let Some(feedback) = self.feedback {
267            feedback.validate(self.continuous.repair_every_source_symbols)?;
268        }
269        let packet_bytes = PACKET_HEADER_LEN
270            .checked_add(usize::from(self.symbol_size))
271            .ok_or(Error::InvalidConfig("resolved MTU overflow"))?;
272        if usize::from(self.mtu_bytes) < packet_bytes {
273            return Err(Error::InvalidConfig("resolved symbol size exceeds the MTU"));
274        }
275        if usize::from(self.symbol_size) <= crate::rlc::FRAGMENT_HEADER_LEN {
276            return Err(Error::InvalidConfig(
277                "MTU leaves no room for continuous record bytes",
278            ));
279        }
280        if usize::from(self.symbol_size) > crate::DEFAULT_MAX_SYMBOL_SIZE
281            || usize::from(self.continuous.window_symbols) > crate::DEFAULT_MAX_WINDOW_SYMBOLS
282        {
283            return Err(Error::InvalidConfig(
284                "resolved FEC values exceed generated sender capacities",
285            ));
286        }
287        if self.bitrate_bps == 0
288            || self.memory_budget_kib == 0
289            || self.max_latency_ms == 0
290            || self.burst_packets == 0
291            || self.continuous.window_symbols == 0
292            || self.continuous.repair_every_source_symbols == 0
293            || self.objects.max_object_bytes == 0
294            || self.objects.repair_symbols_per_block == 0
295            || self.recovery_interval == 0
296            || self.max_record_bytes == 0
297        {
298            return Err(Error::InvalidConfig(
299                "resolved stream plan contains a zero bound",
300            ));
301        }
302        DensityThreshold::new(self.continuous.repair_density)?;
303        let max_record_bytes = usize::try_from(self.max_record_bytes)
304            .map_err(|_| Error::InvalidConfig("record bound exceeds usize"))?;
305        let rlc = self.rlc_config()?;
306        let continuous_bytes = rlc
307            .symbol_size()
308            .checked_mul(rlc.window_symbols())
309            .and_then(|bytes| bytes.checked_add(max_record_bytes))
310            .and_then(|bytes| bytes.checked_add(usize::from(self.mtu_bytes)))
311            .ok_or(Error::InvalidConfig("stream memory estimate overflow"))?;
312        let memory_budget_bytes = usize::try_from(self.memory_budget_kib)
313            .ok()
314            .and_then(|kib| kib.checked_mul(1024))
315            .ok_or(Error::InvalidConfig("stream memory budget overflow"))?;
316        if continuous_bytes > memory_budget_bytes {
317            return Err(Error::InvalidConfig(
318                "continuous sender working set exceeds memory budget",
319            ));
320        }
321        Ok(())
322    }
323
324    pub fn rlc_config(&self) -> Result<RlcConfig> {
325        RlcConfig::new(
326            usize::from(self.symbol_size),
327            usize::from(self.continuous.window_symbols),
328            self.continuous.field.into(),
329        )
330        .map_err(Into::into)
331    }
332
333    /// Selects only the geometry and bounds needed by the receiver.
334    pub fn receiver_requirements(&self) -> ReceiverRequirements {
335        ReceiverRequirements {
336            feedback: self
337                .feedback
338                .map(|policy| crate::feedback::FeedbackRequirements {
339                    report_interval_ms: policy.report_interval_ms,
340                    destination: crate::feedback::destination_key(&self.destination_id),
341                }),
342            symbol_size: self.symbol_size,
343            field: self.continuous.field,
344            window_symbols: self.continuous.window_symbols,
345            max_record_bytes: self.max_record_bytes,
346        }
347    }
348
349    /// Builds both sender lanes and their canonical repeated manifest.
350    pub fn sender_config(
351        &self,
352        identity: StreamIdentity,
353        schema: ApplicationSchema,
354        context: SectionContext,
355    ) -> Result<LogStreamSenderConfig> {
356        self.validate()?;
357        if context.instance_id != identity.sender_id {
358            return Err(Error::InvalidConfig(
359                "manifest instance differs from sender identity",
360            ));
361        }
362        let max_record_bytes = usize::try_from(self.max_record_bytes)
363            .map_err(|_| Error::InvalidConfig("record bound exceeds usize"))?;
364        let manifest =
365            SessionManifest::new(identity, self.receiver_requirements(), schema, context)
366                .encode_record()?;
367        if manifest.len() as u64 > self.objects.max_object_bytes {
368            return Err(Error::ObjectTooLarge {
369                actual: manifest.len() as u64,
370                maximum: self.objects.max_object_bytes,
371            });
372        }
373        Ok(LogStreamSenderConfig {
374            feedback: self.feedback,
375            pacing: crate::PacingConfig {
376                bitrate_bps: self.bitrate_bps,
377                burst_packets: self.burst_packets,
378                max_latency: cu29_clock::CuDuration::from_millis(u64::from(self.max_latency_ms)),
379                memory_budget_bytes: self.memory_budget_kib as usize * 1024,
380            },
381            continuous: ContinuousSenderConfig {
382                identity,
383                lane: Lane::ReplayCritical,
384                fec: self.rlc_config()?,
385                max_record_bytes,
386                initial_esi: EncodingSymbolId::new(0),
387                repair_every_source_symbols: usize::from(
388                    self.continuous.repair_every_source_symbols,
389                ),
390                first_repair_key: 0,
391                repair_density: DensityThreshold::new(self.continuous.repair_density)?,
392            },
393            recovery: RecoverySenderConfig {
394                finite: FiniteObjectSenderConfig {
395                    identity,
396                    lane: Lane::Control,
397                    symbol_size: self.symbol_size,
398                    max_object_bytes: self.objects.max_object_bytes,
399                    repair_symbols_per_block: self.objects.repair_symbols_per_block,
400                },
401                manifest_record: manifest,
402                recovery_interval: self.recovery_interval,
403            },
404        })
405    }
406}
407
408impl From<ResolvedRlcField> for Field {
409    fn from(field: ResolvedRlcField) -> Self {
410        match field {
411            ResolvedRlcField::Gf2 => Self::Gf2,
412            ResolvedRlcField::Gf256 => Self::Gf256,
413        }
414    }
415}
416
417/// Creates a process-local session identity without adding configuration or environment inputs.
418#[cfg(feature = "std")]
419pub fn new_session_id() -> [u8; 16] {
420    use core::sync::atomic::{AtomicU64, Ordering};
421    use std::time::{SystemTime, UNIX_EPOCH};
422
423    static NEXT_SESSION: AtomicU64 = AtomicU64::new(0);
424    let counter = NEXT_SESSION.fetch_add(1, Ordering::Relaxed);
425    let now = SystemTime::now()
426        .duration_since(UNIX_EPOCH)
427        .unwrap_or_default()
428        .as_nanos();
429    let mut hasher = blake3::Hasher::new();
430    hasher.update(&now.to_le_bytes());
431    hasher.update(&counter.to_le_bytes());
432    hasher.update(&std::process::id().to_le_bytes());
433    hasher.finalize().as_bytes()[..16].try_into().unwrap()
434}
cu29_logstream/manifest.rs

cu29_logstream/
manifest.rs
