cu29_unifiedlog/
lib.rs

1#![cfg_attr(not(feature = "std"), no_std)]
2
3extern crate alloc;
4extern crate core;
5
6#[doc(hidden)]
7pub mod byte_log;
8#[cfg(feature = "std")]
9pub mod memmap;
10pub mod noop;
11
12#[cfg(feature = "std")]
13mod compat {
14    // backward compatibility for the std implementation
15    pub use crate::memmap::LogPosition;
16    pub use crate::memmap::MmapUnifiedLogger as UnifiedLogger;
17    pub use crate::memmap::MmapUnifiedLoggerBuilder as UnifiedLoggerBuilder;
18    pub use crate::memmap::MmapUnifiedLoggerRead as UnifiedLoggerRead;
19    pub use crate::memmap::MmapUnifiedLoggerWrite as UnifiedLoggerWrite;
20    pub use crate::memmap::UnifiedLoggerIOReader;
21}
22
23#[cfg(feature = "std")]
24pub use compat::*;
25pub use noop::{NoopLogger, NoopSectionStorage};
26
27use alloc::string::ToString;
28#[cfg(not(feature = "std"))]
29use alloc::sync::Arc;
30use alloc::vec::Vec;
31use core::fmt::{Debug, Display, Formatter, Result as FmtResult};
32#[cfg(not(feature = "std"))]
33use spin::Mutex;
34#[cfg(feature = "std")]
35use std::sync::{Arc, Mutex};
36
37use bincode::error::EncodeError;
38use bincode::{Decode, Encode};
39use cu29_traits::{CuError, CuResult, UnifiedLogType, WriteStream};
40
41/// ID to spot the beginning of a Copper Log
42#[allow(dead_code)]
43pub const MAIN_MAGIC: [u8; 4] = [0xB4, 0xA5, 0x50, 0xFF]; // BRASS OFF
44
45/// ID to spot a section of Copper Log
46pub const SECTION_MAGIC: [u8; 2] = [0xFA, 0x57]; // FAST
47
48/// Version of the unified log **encapsulation only**: file headers, section
49/// headers, and the layout used to locate sections in a slab.
50///
51/// This is **never** a version of the encoded content inside sections. Changes
52/// to CopperLists, payload types, keyframes, or their serialization must not bump
53/// this value. Decode content with the logreader built for the exact application
54/// version that produced it; this header cannot establish content compatibility.
55/// Unreleased version 2 uses byte links, static metadata and construction contexts.
56pub const UNIFIED_LOG_FORMAT_VERSION: u8 = 2;
57
58pub const SECTION_HEADER_COMPACT_SIZE: u16 = 512; // Usual minimum size for a disk sector.
59
60/// Header of the byte-addressed log. All offsets are relative to the log origin.
61#[derive(Encode, Decode, Debug, Clone)]
62pub struct MainHeader {
63    pub magic: [u8; 4],
64    pub format_version: u8,
65    pub page_size: u16,
66    pub metadata_offset: u64,
67    pub sections_begin: u64,
68    pub sections_end: u64,
69    pub head_section: u64,
70    pub tail_section: u64,
71    pub clean_close: bool,
72}
73
74impl Display for MainHeader {
75    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
76        write!(
77            f,
78            "  format_version -> {}\n  alignment -> {}\n  metadata -> {}\n  data -> {}..{}\n  head -> {}\n  tail -> {}\n  clean_close -> {}",
79            self.format_version,
80            self.page_size,
81            self.metadata_offset,
82            self.sections_begin,
83            self.sections_end,
84            self.head_section,
85            self.tail_section,
86            self.clean_close
87        )
88    }
89}
90
91/// Construction identity shared by every section of an application instance.
92#[derive(Encode, Decode, Debug, Clone, Copy, Default, PartialEq, Eq)]
93pub struct SectionContext {
94    pub run_id: u64,
95    pub instance_id: u32,
96    pub mission_index: u32,
97}
98
99/// One linked section. The allocation includes its 512-byte header reservation.
100#[derive(Encode, Decode, Debug, Clone)]
101pub struct SectionHeader {
102    pub magic: [u8; 2],
103    pub block_size: u16,
104    pub entry_type: UnifiedLogType,
105    pub allocated: u64,
106    pub next_section: u64,
107    pub used: u32,
108    pub is_open: bool,
109    pub context: SectionContext,
110}
111
112impl Display for SectionHeader {
113    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
114        write!(
115            f,
116            "    type -> {:?}\n    use -> {} / {} (open: {})\n    next -> {}\n    context -> {:?}",
117            self.entry_type,
118            self.used,
119            self.allocated,
120            self.is_open,
121            self.next_section,
122            self.context
123        )
124    }
125}
126
127impl Default for SectionHeader {
128    fn default() -> Self {
129        Self {
130            magic: SECTION_MAGIC,
131            block_size: SECTION_HEADER_COMPACT_SIZE,
132            entry_type: UnifiedLogType::Empty,
133            allocated: 0,
134            next_section: 0,
135            used: 0,
136            is_open: true,
137            context: SectionContext::default(),
138        }
139    }
140}
141
142/// Static application identity and configuration, retained outside the data ring.
143#[derive(Encode, Decode, Debug, Clone, PartialEq, Eq)]
144pub struct ApplicationMetadata {
145    pub app_type: alloc::string::String,
146    pub app_name: alloc::string::String,
147    pub app_version: alloc::string::String,
148    pub git_commit: Option<alloc::string::String>,
149    pub git_dirty: Option<bool>,
150    pub subsystem_id: Option<alloc::string::String>,
151    pub subsystem_code: u16,
152    pub effective_config_ron: alloc::string::String,
153    pub missions: Vec<alloc::string::String>,
154    pub catalog_offset: u64,
155}
156
157/// Behavior when a section needs additional storage.
158#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
159pub enum CapacityPolicy {
160    #[default]
161    Grow,
162    OverwriteOldest,
163    StopWhenFull,
164}
165
166pub enum AllocatedSection<S: SectionStorage> {
167    NoMoreSpace,
168    Section(SectionHandle<S>),
169}
170
171/// A Storage is an append-only structure that can update a header section.
172pub trait SectionStorage: Send + Sync {
173    /// This rewinds the storage, serialize the header and jumps to the beginning of the user data storage.
174    fn initialize<E: Encode>(&mut self, header: &E) -> Result<usize, EncodeError>;
175    /// This updates the header leaving the position to the end of the user data storage.
176    fn post_update_header<E: Encode>(&mut self, header: &E) -> Result<usize, EncodeError>;
177    /// Appends the entry to the user data storage.
178    fn append<E: Encode>(&mut self, entry: &E) -> Result<usize, EncodeError>;
179    /// Whether this handle was sealed by section reclamation.
180    fn is_sealed(&self) -> bool {
181        false
182    }
183    /// Flushes the section to the underlying storage
184    fn flush(&mut self) -> CuResult<usize>;
185}
186
187/// A SectionHandle is a handle to a section in the datalogger.
188/// It allows tracking the lifecycle of the section.
189#[derive(Default)]
190pub struct SectionHandle<S: SectionStorage> {
191    header: SectionHeader, // keep a copy of the header as metadata
192    storage: S,
193}
194
195impl<S: SectionStorage> SectionHandle<S> {
196    pub fn create(header: SectionHeader, mut storage: S) -> CuResult<Self> {
197        // Write the first version of the header.
198        let _ = storage.initialize(&header).map_err(|e| e.to_string())?;
199        Ok(Self { header, storage })
200    }
201
202    pub fn mark_closed(&mut self) {
203        self.header.is_open = false;
204    }
205    pub fn append<E: Encode>(&mut self, entry: E) -> Result<usize, EncodeError> {
206        self.storage.append(&entry)
207    }
208
209    pub fn get_storage(&self) -> &S {
210        &self.storage
211    }
212
213    pub fn get_storage_mut(&mut self) -> &mut S {
214        &mut self.storage
215    }
216
217    pub fn post_update_header(&mut self) -> Result<usize, EncodeError> {
218        self.storage.post_update_header(&self.header)
219    }
220}
221
222/// Basic statistics for the unified logger.
223/// Note: the total_allocated_space might grow for the std implementation
224pub struct UnifiedLogStatus {
225    /// Bytes reserved for headers, static metadata and retained data sections.
226    pub total_used_space: usize,
227    /// Total logical backing capacity, including free space.
228    pub total_allocated_space: usize,
229}
230
231/// Payload stored in the end-of-log section to signal whether the log was cleanly closed.
232#[derive(Encode, Decode, Debug, Clone)]
233pub struct EndOfLogMarker {
234    pub temporary: bool,
235}
236
237/// The writing interface to the unified logger.
238/// Writing is "almost" linear as various streams can allocate sections and track them until
239/// they drop them.
240pub trait UnifiedLogWrite<S: SectionStorage>: Send + Sync {
241    /// A section is a contiguous chunk of memory that can be used to write data.
242    /// It can store various types of data as specified by the entry_type.
243    /// The requested_section_size is the size of the section to allocate.
244    /// It returns a handle to the section that can be used to write data until
245    /// it is flushed with flush_section, it is then considered unmutable.
246    fn add_section(
247        &mut self,
248        entry_type: UnifiedLogType,
249        requested_section_size: usize,
250    ) -> CuResult<SectionHandle<S>>;
251
252    /// Flush the given section to the underlying storage.
253    fn flush_section(&mut self, section: &mut SectionHandle<S>);
254
255    /// Flush a section, reporting storage failures.
256    fn try_flush_section(&mut self, section: &mut SectionHandle<S>) -> CuResult<()> {
257        self.flush_section(section);
258        Ok(())
259    }
260
261    /// Seal or compare static metadata before initializing runtime streams.
262    #[doc(hidden)]
263    fn seal_metadata<C: Encode>(
264        &mut self,
265        _metadata: &ApplicationMetadata,
266        _catalog: Option<&C>,
267    ) -> CuResult<()> {
268        Err(CuError::from(
269            "Logger does not support application metadata",
270        ))
271    }
272    /// Reserve a construction identity after static metadata has been compared.
273    #[doc(hidden)]
274    fn construction_context(
275        &mut self,
276        instance_id: u32,
277        mission_index: u32,
278    ) -> CuResult<SectionContext> {
279        Ok(SectionContext {
280            run_id: 0,
281            instance_id,
282            mission_index,
283        })
284    }
285    /// Allocate a section belonging to one construction.
286    #[doc(hidden)]
287    fn add_section_with_context(
288        &mut self,
289        kind: UnifiedLogType,
290        size: usize,
291        _context: SectionContext,
292    ) -> CuResult<SectionHandle<S>> {
293        self.add_section(kind, size)
294    }
295    /// Returns the current status of the unified logger.
296    fn status(&self) -> UnifiedLogStatus;
297}
298
299/// Read back a unified log linearly.
300pub trait UnifiedLogRead {
301    /// Read through the unified logger until it reaches the UnifiedLogType given in datalogtype.
302    /// It will return the byte array of the section if found.
303    fn read_next_section_type(&mut self, datalogtype: UnifiedLogType) -> CuResult<Option<Vec<u8>>>;
304
305    /// Read through the next section entry regardless of its type.
306    /// It will return the header and the byte array of the section.
307    /// Note the last Entry should be of UnifiedLogType::LastEntry if the log is not corrupted.
308    fn raw_read_section(&mut self) -> CuResult<(SectionHeader, Vec<u8>)>;
309}
310
311/// Create a new stream to write to the unifiedlogger.
312pub fn stream_write<E: Encode, S: SectionStorage>(
313    logger: Arc<Mutex<impl UnifiedLogWrite<S>>>,
314    entry_type: UnifiedLogType,
315    minimum_allocation_amount: usize,
316) -> CuResult<impl WriteStream<E>> {
317    LogStream::new(entry_type, logger, minimum_allocation_amount)
318}
319
320/// Create a stream whose sections retain their construction identity across rollover.
321#[doc(hidden)]
322pub fn stream_write_context<E: Encode, S: SectionStorage>(
323    logger: Arc<Mutex<impl UnifiedLogWrite<S>>>,
324    entry_type: UnifiedLogType,
325    size: usize,
326    context: SectionContext,
327) -> CuResult<impl WriteStream<E>> {
328    LogStream::with_context(entry_type, logger, size, context)
329}
330
331/// A wrapper around the unifiedlogger that implements the Write trait.
332pub struct LogStream<S: SectionStorage, L: UnifiedLogWrite<S>> {
333    entry_type: UnifiedLogType,
334    parent_logger: Arc<Mutex<L>>,
335    current_section: SectionHandle<S>,
336    current_position: usize,
337    minimum_allocation_amount: usize,
338    last_log_bytes: usize,
339    context: SectionContext,
340}
341
342impl<S: SectionStorage, L: UnifiedLogWrite<S>> LogStream<S, L> {
343    /// Creates a concrete stream, including for borrowed canonical encoded entries.
344    pub fn new(
345        entry_type: UnifiedLogType,
346        parent_logger: Arc<Mutex<L>>,
347        minimum_allocation_amount: usize,
348    ) -> CuResult<Self> {
349        Self::with_context(
350            entry_type,
351            parent_logger,
352            minimum_allocation_amount,
353            SectionContext::default(),
354        )
355    }
356
357    #[doc(hidden)]
358    pub fn with_context(
359        entry_type: UnifiedLogType,
360        parent_logger: Arc<Mutex<L>>,
361        minimum_allocation_amount: usize,
362        context: SectionContext,
363    ) -> CuResult<Self> {
364        #[cfg(feature = "std")]
365        let section = parent_logger
366            .lock()
367            .map_err(|e| {
368                CuError::from("Could not lock a section at LogStream creation")
369                    .add_cause(e.to_string().as_str())
370            })?
371            .add_section_with_context(entry_type, minimum_allocation_amount, context)?;
372
373        #[cfg(not(feature = "std"))]
374        let section = parent_logger.lock().add_section_with_context(
375            entry_type,
376            minimum_allocation_amount,
377            context,
378        )?;
379
380        Ok(Self {
381            entry_type,
382            parent_logger,
383            current_section: section,
384            current_position: 0,
385            minimum_allocation_amount,
386            last_log_bytes: 0,
387            context,
388        })
389    }
390}
391
392impl<S: SectionStorage, L: UnifiedLogWrite<S>> Debug for LogStream<S, L> {
393    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
394        write!(
395            f,
396            "MmapStream {{ entry_type: {:?}, current_position: {}, minimum_allocation_amount: {} }}",
397            self.entry_type, self.current_position, self.minimum_allocation_amount
398        )
399    }
400}
401
402impl<E: Encode, S: SectionStorage, L: UnifiedLogWrite<S>> WriteStream<E> for LogStream<S, L> {
403    fn log(&mut self, obj: &E) -> CuResult<()> {
404        let sealed = self.current_section.storage.is_sealed();
405        let result = if sealed {
406            Err(EncodeError::UnexpectedEnd)
407        } else {
408            self.current_section.append(obj)
409        };
410        match result {
411            Ok(nb_bytes) => {
412                self.current_position += nb_bytes;
413                self.current_section.header.used += nb_bytes as u32;
414                self.last_log_bytes = nb_bytes;
415                // Lifecycle sections are promptly closed, so sparse events cannot pin the ring.
416                if self.entry_type == UnifiedLogType::RuntimeLifecycle {
417                    self.current_section.storage.flush()?;
418                }
419                Ok(())
420            }
421            Err(e) => match e {
422                EncodeError::UnexpectedEnd => {
423                    if !sealed && self.current_section.header.used == 0 {
424                        return Err(CuError::from("Entry exceeds an empty section"));
425                    }
426                    #[cfg(feature = "std")]
427                    let logger_guard = self.parent_logger.lock();
428
429                    #[cfg(not(feature = "std"))]
430                    let mut logger_guard = self.parent_logger.lock();
431
432                    #[cfg(feature = "std")]
433                    let mut logger_guard =
434                        match logger_guard {
435                            Ok(g) => g,
436                            Err(_) => return Err(
437                                "Logger mutex poisoned while reporting EncodeError::UnexpectedEnd"
438                                    .into(),
439                            ), // It will retry but at least not completely crash.
440                        };
441
442                    logger_guard.try_flush_section(&mut self.current_section)?;
443                    self.current_section = logger_guard.add_section_with_context(
444                        self.entry_type,
445                        self.minimum_allocation_amount,
446                        self.context,
447                    )?;
448
449                    let result = self
450                        .current_section
451                        .append(obj)
452                        .map_err(|e| {
453                            CuError::from(
454                                "Failed to encode object in a newly minted section. Unrecoverable failure.",
455                            )
456                            .add_cause(e.to_string().as_str())
457                        })?; // If we fail just after creating a section, there is not much we can do.
458
459                    self.current_position += result;
460                    self.current_section.header.used += result as u32;
461                    self.last_log_bytes = result;
462                    if self.entry_type == UnifiedLogType::RuntimeLifecycle {
463                        self.current_section.storage.flush()?;
464                    }
465                    Ok(())
466                }
467                _ => {
468                    let err =
469                        <&str as Into<CuError>>::into("Unexpected error while encoding object.")
470                            .add_cause(e.to_string().as_str());
471                    Err(err)
472                }
473            },
474        }
475    }
476
477    fn flush(&mut self) -> CuResult<()> {
478        #[cfg(feature = "std")]
479        let mut logger = self
480            .parent_logger
481            .lock()
482            .map_err(|_| CuError::from("Logger mutex poisoned"))?;
483        #[cfg(not(feature = "std"))]
484        let mut logger = self.parent_logger.lock();
485        logger.try_flush_section(&mut self.current_section)
486    }
487
488    fn last_log_bytes(&self) -> Option<usize> {
489        Some(self.last_log_bytes)
490    }
491}
492
493impl<S: SectionStorage, L: UnifiedLogWrite<S>> Drop for LogStream<S, L> {
494    fn drop(&mut self) {
495        #[cfg(feature = "std")]
496        match self.parent_logger.lock() {
497            Ok(mut logger_guard) => {
498                logger_guard.flush_section(&mut self.current_section);
499            }
500            Err(_) => {
501                // Only surface the warning when a real poisoning occurred.
502                if !std::thread::panicking() {
503                    eprintln!("⚠️ MmapStream::drop: logger mutex poisoned");
504                }
505            }
506        }
507
508        #[cfg(not(feature = "std"))]
509        {
510            let mut logger_guard = self.parent_logger.lock();
511            logger_guard.flush_section(&mut self.current_section);
512        }
513    }
514}
cu29_unifiedlog/lib.rs

cu29_unifiedlog/
lib.rs
