cu29_runtime/
logcodec.rs

1#[cfg(not(feature = "std"))]
2extern crate alloc;
3
4use crate::config::{ComponentConfig, CuConfig, LoggingCodecSpec};
5use crate::cutask::{CuMsg, CuMsgMetadata, CuMsgPayload};
6use crate::sync_compat::{Mutex, OnceLock, lock as lock_mutex, once_get_or_init};
7use alloc::boxed::Box;
8use alloc::format;
9use alloc::string::{String, ToString};
10use bincode::de::{Decode, Decoder};
11use bincode::enc::{Encode, Encoder};
12use bincode::error::{DecodeError, EncodeError};
13use core::any::TypeId;
14use cu29_clock::Tov;
15use cu29_traits::{CuError, CuResult, observed_encode_bytes};
16use hashbrown::HashMap;
17use portable_atomic::{AtomicU64, Ordering};
18use serde::de::DeserializeOwned;
19#[cfg(feature = "std")]
20use std::path::Path;
21
22#[cfg(feature = "std")]
23use cu29_unifiedlog::{UnifiedLogger, UnifiedLoggerBuilder};
24
25pub trait CuLogCodec<P: CuMsgPayload>: 'static {
26    type Config: DeserializeOwned + Default;
27
28    fn new(config: Self::Config) -> CuResult<Self>
29    where
30        Self: Sized;
31
32    /// Returns handle-backed source bytes read directly by the codec.
33    ///
34    /// This reports only extra handle-backed residency beyond the payload's
35    /// fixed `size_of::<P>()` footprint already accounted by runtime
36    /// monitoring. Codecs must implement this explicitly so they opt into the
37    /// correct accounting model for their payload type.
38    fn source_payload_handle_bytes(&self, payload: &P) -> usize;
39
40    fn encode_payload<E: Encoder>(
41        &mut self,
42        payload: &P,
43        encoder: &mut E,
44    ) -> Result<(), EncodeError>;
45
46    fn decode_payload<D: Decoder<Context = ()>>(
47        &mut self,
48        decoder: &mut D,
49    ) -> Result<P, DecodeError>;
50}
51
52pub struct CodecState<C> {
53    inner: Mutex<Option<(u64, C)>>,
54}
55
56impl<C> CodecState<C> {
57    pub const fn new() -> Self {
58        Self {
59            inner: Mutex::new(None),
60        }
61    }
62}
63
64impl<C> Default for CodecState<C> {
65    fn default() -> Self {
66        Self::new()
67    }
68}
69
70pub struct EffectiveConfigEntry {
71    version: AtomicU64,
72    ron: Mutex<String>,
73}
74
75impl EffectiveConfigEntry {
76    fn new(ron: &str) -> Self {
77        Self {
78            version: AtomicU64::new(1),
79            ron: Mutex::new(ron.to_string()),
80        }
81    }
82
83    pub fn version(&self) -> u64 {
84        self.version.load(Ordering::Acquire)
85    }
86
87    pub fn ron(&self) -> String {
88        lock_mutex(&self.ron).clone()
89    }
90
91    fn set(&self, ron: &str) {
92        *lock_mutex(&self.ron) = ron.to_string();
93        self.version.fetch_add(1, Ordering::AcqRel);
94    }
95}
96
97type EffectiveConfigRegistry = HashMap<TypeId, &'static EffectiveConfigEntry>;
98
99static EFFECTIVE_CONFIGS: OnceLock<Mutex<EffectiveConfigRegistry>> = OnceLock::new();
100
101fn effective_config_registry() -> &'static Mutex<EffectiveConfigRegistry> {
102    once_get_or_init(&EFFECTIVE_CONFIGS, || Mutex::new(HashMap::new()))
103}
104
105pub fn effective_config_entry<T: 'static>(default_ron: &str) -> &'static EffectiveConfigEntry {
106    let registry = effective_config_registry();
107    let mut registry = lock_mutex(registry);
108    if let Some(entry) = registry.get(&TypeId::of::<T>()) {
109        return entry;
110    }
111
112    let entry = Box::leak(Box::new(EffectiveConfigEntry::new(default_ron)));
113    registry.insert(TypeId::of::<T>(), entry);
114    entry
115}
116
117pub fn set_effective_config_ron<T: 'static>(ron: &str) {
118    effective_config_entry::<T>(ron).set(ron);
119}
120
121pub fn with_codec_for_encode<C, R, B, F>(
122    state: &'static CodecState<C>,
123    config_entry: &EffectiveConfigEntry,
124    build: B,
125    f: F,
126) -> Result<R, EncodeError>
127where
128    B: FnOnce(&str) -> CuResult<C>,
129    F: FnOnce(&mut C) -> Result<R, EncodeError>,
130{
131    let version = config_entry.version();
132    let mut guard = lock_mutex(&state.inner);
133    if guard
134        .as_ref()
135        .is_none_or(|(cached_version, _)| *cached_version != version)
136    {
137        let effective_config_ron = config_entry.ron();
138        let codec = build(&effective_config_ron)
139            .map_err(|err| EncodeError::OtherString(err.to_string()))?;
140        *guard = Some((version, codec));
141    }
142    let (_, codec) = guard
143        .as_mut()
144        .expect("codec state must be initialized after build");
145    f(codec)
146}
147
148pub fn with_codec_for_decode<C, R, B, F>(
149    state: &'static CodecState<C>,
150    config_entry: &EffectiveConfigEntry,
151    build: B,
152    f: F,
153) -> Result<R, DecodeError>
154where
155    B: FnOnce(&str) -> CuResult<C>,
156    F: FnOnce(&mut C) -> Result<R, DecodeError>,
157{
158    let version = config_entry.version();
159    let mut guard = lock_mutex(&state.inner);
160    if guard
161        .as_ref()
162        .is_none_or(|(cached_version, _)| *cached_version != version)
163    {
164        let effective_config_ron = config_entry.ron();
165        let codec = build(&effective_config_ron)
166            .map_err(|err| DecodeError::OtherString(err.to_string()))?;
167        *guard = Some((version, codec));
168    }
169    let (_, codec) = guard
170        .as_mut()
171        .expect("codec state must be initialized after build");
172    f(codec)
173}
174
175pub fn resolve_task_output_codec<'a>(
176    config: &'a CuConfig,
177    mission_id: Option<&str>,
178    task_id: &str,
179    msg_type: &str,
180) -> CuResult<Option<&'a LoggingCodecSpec>> {
181    let node = config.find_task_node(mission_id, task_id).ok_or_else(|| {
182        CuError::from(format!(
183            "Could not find task '{task_id}' while resolving log codec for '{msg_type}'."
184        ))
185    })?;
186
187    let codec_id = node
188        .get_logging()
189        .and_then(|logging| logging.codec_for_msg_type(msg_type));
190    let Some(codec_id) = codec_id else {
191        return Ok(None);
192    };
193
194    config
195        .find_logging_codec_spec(codec_id)
196        .map(Some)
197        .ok_or_else(|| {
198            CuError::from(format!(
199                "Task '{task_id}' binds output '{msg_type}' to unknown logging codec '{codec_id}'."
200            ))
201        })
202}
203
204pub fn instantiate_codec<C, P>(
205    effective_config_ron: &str,
206    mission_id: Option<&str>,
207    task_id: &str,
208    msg_type: &str,
209    expected_type_path: &str,
210) -> CuResult<C>
211where
212    C: CuLogCodec<P>,
213    P: CuMsgPayload,
214{
215    let config = CuConfig::deserialize_ron(effective_config_ron)?;
216    let spec = resolve_task_output_codec(&config, mission_id, task_id, msg_type)?.ok_or_else(
217        || {
218            CuError::from(format!(
219                "Task '{task_id}' output '{msg_type}' has no configured logging codec in the effective config."
220            ))
221        },
222    )?;
223
224    if spec.type_ != expected_type_path {
225        return Err(CuError::from(format!(
226            "Task '{task_id}' output '{msg_type}' resolved logging codec type '{}' but '{}' was compiled for this slot.",
227            spec.type_, expected_type_path
228        )));
229    }
230
231    let codec_config = deserialize_codec_config::<C, P>(spec.config.as_ref())?;
232    C::new(codec_config)
233}
234
235pub fn deserialize_codec_config<C, P>(config: Option<&ComponentConfig>) -> CuResult<C::Config>
236where
237    C: CuLogCodec<P>,
238    P: CuMsgPayload,
239{
240    match config {
241        Some(config) => config.deserialize_into::<C::Config>().map_err(|err| {
242            CuError::from(format!(
243                "Failed to deserialize logging codec config for payload '{}': {err}",
244                core::any::type_name::<P>()
245            ))
246        }),
247        None => Ok(C::Config::default()),
248    }
249}
250
251pub fn encode_msg_with_codec<T, C, E>(
252    msg: &CuMsg<T>,
253    codec: &mut C,
254    encoder: &mut E,
255) -> Result<(), EncodeError>
256where
257    T: CuMsgPayload,
258    C: CuLogCodec<T>,
259    E: Encoder,
260{
261    // NOTE: like `Encode for CuStampedData` in cutask.rs, this is generic over the
262    // payload type, so it cannot itself consult `HandleContent` policy via the
263    // autoref-specialization pattern. The codegen-emitted per-slot encoder calls
264    // `cu29::cutask::encode_metadata_only` instead when bytes should be skipped.
265    match msg.payload() {
266        None => {
267            0u8.encode(encoder)?;
268        }
269        Some(payload) => {
270            1u8.encode(encoder)?;
271            let encoded_start = observed_encode_bytes();
272            let handle_start = crate::monitoring::current_payload_handle_bytes();
273            let source_handle_bytes = codec.source_payload_handle_bytes(payload);
274            if source_handle_bytes > 0 {
275                crate::monitoring::record_payload_handle_bytes(source_handle_bytes);
276            }
277            codec.encode_payload(payload, encoder)?;
278            let encoded_bytes = observed_encode_bytes().saturating_sub(encoded_start);
279            let handle_bytes =
280                crate::monitoring::current_payload_handle_bytes().saturating_sub(handle_start);
281            crate::monitoring::record_current_slot_payload_io_stats(
282                core::mem::size_of::<T>(),
283                encoded_bytes,
284                handle_bytes,
285            );
286        }
287    }
288    msg.tov.encode(encoder)?;
289    msg.metadata.encode(encoder)?;
290    Ok(())
291}
292
293pub fn decode_msg_with_codec<T, C, D>(
294    decoder: &mut D,
295    codec: &mut C,
296) -> Result<CuMsg<T>, DecodeError>
297where
298    T: CuMsgPayload,
299    C: CuLogCodec<T>,
300    D: Decoder<Context = ()>,
301{
302    let present: u8 = Decode::decode(decoder)?;
303    let payload = match present {
304        0 => None,
305        1 => Some(codec.decode_payload(decoder)?),
306        value => {
307            return Err(DecodeError::OtherString(format!(
308                "Invalid CuMsg presence tag {value} for payload '{}'",
309                core::any::type_name::<T>()
310            )));
311        }
312    };
313    let tov: Tov = Decode::decode(decoder)?;
314    let metadata: CuMsgMetadata = Decode::decode(decoder)?;
315    Ok(CuMsg::from_parts(payload, tov, metadata))
316}
317
318/// Encodes one present payload with its configured codec. Presence and common
319/// metadata are carried by the generated CopperList metadata block.
320#[doc(hidden)]
321pub fn encode_payload_with_codec<T, C, E>(
322    payload: &T,
323    codec: &mut C,
324    encoder: &mut E,
325) -> Result<(), EncodeError>
326where
327    T: CuMsgPayload,
328    C: CuLogCodec<T>,
329    E: Encoder,
330{
331    let encoded_start = observed_encode_bytes();
332    let handle_start = crate::monitoring::current_payload_handle_bytes();
333    let source_handle_bytes = codec.source_payload_handle_bytes(payload);
334    if source_handle_bytes > 0 {
335        crate::monitoring::record_payload_handle_bytes(source_handle_bytes);
336    }
337    codec.encode_payload(payload, encoder)?;
338    let encoded_bytes = observed_encode_bytes().saturating_sub(encoded_start);
339    let handle_bytes =
340        crate::monitoring::current_payload_handle_bytes().saturating_sub(handle_start);
341    crate::monitoring::record_current_slot_payload_io_stats(
342        core::mem::size_of::<T>(),
343        encoded_bytes,
344        handle_bytes,
345    );
346    Ok(())
347}
348
349/// Decodes one payload with its configured codec after generated metadata has
350/// declared that the payload bytes are present.
351#[doc(hidden)]
352pub fn decode_payload_with_codec<T, C, D>(decoder: &mut D, codec: &mut C) -> Result<T, DecodeError>
353where
354    T: CuMsgPayload,
355    C: CuLogCodec<T>,
356    D: Decoder<Context = ()>,
357{
358    codec.decode_payload(decoder)
359}
360
361#[cfg(feature = "std")]
362pub fn read_effective_config_ron_from_log(log_base: &Path) -> CuResult<Option<String>> {
363    let logger = UnifiedLoggerBuilder::new()
364        .file_base_name(log_base)
365        .build()
366        .map_err(|err| {
367            CuError::new_with_cause(
368                &format!(
369                    "Failed to open Copper log '{}' while loading effective log config",
370                    log_base.display()
371                ),
372                err,
373            )
374        })?;
375    let UnifiedLogger::Read(read_logger) = logger else {
376        return Err(CuError::from(
377            "Expected readable unified logger while loading effective log config",
378        ));
379    };
380
381    Ok(read_logger
382        .application_metadata()?
383        .map(|metadata| metadata.effective_config_ron))
384}
385
386#[cfg(feature = "std")]
387pub fn seed_effective_config_from_log<T: 'static>(log_base: &Path) -> CuResult<Option<String>> {
388    let effective_config_ron = read_effective_config_ron_from_log(log_base)?;
389    if let Some(ref ron) = effective_config_ron {
390        set_effective_config_ron::<T>(ron);
391    }
392    Ok(effective_config_ron)
393}
cu29_runtime/logcodec.rs

cu29_runtime/
logcodec.rs
