cu29_log_runtime/
lib.rs

1#![cfg_attr(not(feature = "std"), no_std)]
2#[cfg(not(feature = "std"))]
3extern crate alloc;
4
5mod sync_compat;
6
7use core::sync::atomic::{AtomicUsize, Ordering};
8use cu29_clock::RobotClock;
9use cu29_log::CuLogEntry;
10#[allow(unused_imports)]
11use cu29_log::CuLogLevel;
12use cu29_traits::{CuResult, WriteStream};
13use log::Log;
14use sync_compat::{Mutex, OnceLock, init_once, lock as lock_mutex};
15
16#[cfg(not(feature = "std"))]
17mod imp {
18    pub use alloc::boxed::Box;
19    pub use alloc::vec::Vec;
20}
21
22#[cfg(feature = "std")]
23mod imp {
24    pub use bincode::config::Configuration;
25    pub use bincode::enc::Encode;
26    pub use bincode::enc::Encoder;
27    pub use bincode::enc::EncoderImpl;
28    pub use bincode::enc::write::Writer;
29    pub use bincode::error::EncodeError;
30    pub use std::fmt::{Debug, Formatter};
31    pub use std::fs::File;
32    pub use std::io::{BufWriter, Write};
33    pub use std::path::PathBuf;
34
35    #[cfg(debug_assertions)]
36    pub use {std::collections::HashMap, strfmt::strfmt};
37}
38
39use imp::*;
40
41#[allow(dead_code)] // for no_std
42#[derive(Debug)]
43struct DummyWriteStream;
44
45impl WriteStream<CuLogEntry> for DummyWriteStream {
46    #[allow(unused_variables)] // for no_std
47    fn log(&mut self, obj: &CuLogEntry) -> CuResult<()> {
48        #[cfg(feature = "std")]
49        eprintln!("Pending logs got cut: {obj:?}");
50        Ok(())
51    }
52}
53/// Existing type-erased leaf boundary for structured-log consumers.
54///
55/// Generated code may place a statically composed semantic-output adapter here
56/// without adding another layer of runtime dispatch.
57type StructuredLogSink = Box<dyn WriteStream<CuLogEntry> + Send + 'static>;
58
59/// Callback signature: receives the structured entry plus its format string and param names.
60pub type LiveLogListener = Box<dyn Fn(&CuLogEntry, &str, &[&str]) + Send + Sync + 'static>;
61
62pub type LiveLogListenerId = usize;
63
64struct LiveLogRegistration {
65    id: LiveLogListenerId,
66    callback: LiveLogListener,
67    #[cfg(all(feature = "std", debug_assertions))]
68    output: LiveLogOutput,
69}
70
71#[cfg(all(feature = "std", debug_assertions))]
72#[derive(Clone, Copy, PartialEq, Eq)]
73enum LiveLogOutput {
74    Observer,
75    Stdout,
76    Monitor,
77}
78
79struct LiveLogListeners {
80    next_id: LiveLogListenerId,
81    listeners: Vec<LiveLogRegistration>,
82}
83
84impl Default for LiveLogListeners {
85    fn default() -> Self {
86        Self {
87            next_id: 1,
88            listeners: Vec::new(),
89        }
90    }
91}
92
93impl LiveLogListeners {
94    fn insert(&mut self, listener: LiveLogListener) -> LiveLogListenerId {
95        let id = self.next_id;
96        self.next_id = self.next_id.saturating_add(1).max(1);
97        self.listeners.push(LiveLogRegistration {
98            id,
99            callback: listener,
100            #[cfg(all(feature = "std", debug_assertions))]
101            output: LiveLogOutput::Observer,
102        });
103        id
104    }
105
106    fn remove(&mut self, id: LiveLogListenerId) {
107        self.listeners.retain(|listener| listener.id != id);
108    }
109
110    fn notify(&self, entry: &CuLogEntry, format_str: &str, param_names: &[&str]) {
111        #[cfg(all(feature = "std", debug_assertions))]
112        let monitor_active = self
113            .listeners
114            .iter()
115            .any(|listener| listener.output == LiveLogOutput::Monitor);
116        #[cfg(all(feature = "std", debug_assertions))]
117        let mut stdout_notified = false;
118        for listener in &self.listeners {
119            #[cfg(all(feature = "std", debug_assertions))]
120            if listener.output == LiveLogOutput::Stdout {
121                if monitor_active || stdout_notified {
122                    continue;
123                }
124                stdout_notified = true;
125            }
126            (listener.callback)(entry, format_str, param_names);
127        }
128    }
129}
130
131pub struct LiveLogListenerGuard {
132    id: Option<LiveLogListenerId>,
133}
134
135impl Drop for LiveLogListenerGuard {
136    fn drop(&mut self) {
137        if let Some(id) = self.id.take() {
138            unregister_live_log_listener_id(id);
139        }
140    }
141}
142
143#[cfg(all(feature = "std", debug_assertions))]
144pub fn format_message_only(
145    format_str: &str,
146    params: &[String],
147    named_params: &HashMap<String, String>,
148) -> CuResult<String> {
149    if format_str.contains("{}") {
150        let mut formatted = format_str.to_string();
151        for param in params.iter() {
152            if !formatted.contains("{}") {
153                break;
154            }
155            formatted = formatted.replacen("{}", param, 1);
156        }
157        if !named_params.is_empty() {
158            let mut named = named_params.iter().collect::<Vec<_>>();
159            named.sort_by(|a, b| a.0.cmp(b.0));
160            for (name, value) in named {
161                if formatted.contains("{}") {
162                    formatted = formatted.replacen("{}", value, 1);
163                }
164                formatted = formatted.replace(&format!("{{{name}}}"), value);
165            }
166        }
167        return Ok(formatted);
168    }
169
170    // Named replacement
171    imp::strfmt(format_str, named_params).map_err(|e| {
172        cu29_traits::CuError::new_with_cause(
173            format!(
174                "Failed to format log message: {format_str:?} with variables [{named_params:?}]"
175            )
176            .as_str(),
177            e,
178        )
179    })
180}
181
182/// Shared logging state reachable from the macro-generated calls.
183struct LoggerState {
184    sink: Mutex<StructuredLogSink>,
185    clock: RobotClock,
186    live_listeners: Mutex<LiveLogListeners>,
187}
188
189impl core::fmt::Debug for LoggerState {
190    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
191        f.debug_struct("LoggerState")
192            .field("clock", &self.clock)
193            .finish_non_exhaustive()
194    }
195}
196
197static LOGGER_STATE: OnceLock<LoggerState> = OnceLock::new();
198static STRUCTURED_LOG_BYTES: AtomicUsize = AtomicUsize::new(0);
199
200fn init_logger_state(state: LoggerState) {
201    init_once(&LOGGER_STATE, state);
202}
203
204pub struct NullLog;
205impl Log for NullLog {
206    fn enabled(&self, _metadata: &log::Metadata) -> bool {
207        false
208    }
209
210    fn log(&self, _record: &log::Record) {}
211    fn flush(&self) {}
212}
213
214/// The lifetime of this struct is the lifetime of the logger.
215pub struct LoggerRuntime {}
216
217impl LoggerRuntime {
218    /// `sink` consumes each semantic structured-log entry.
219    ///
220    /// The default generated sink writes the local binary stream; generated
221    /// applications may instead supply a statically composed output adapter.
222    /// `extra_text_logger` is the logger that will log the text logs in real time. This is slow and only for debug builds.
223    pub fn init(
224        clock: RobotClock,
225        sink: impl WriteStream<CuLogEntry> + 'static,
226        #[allow(unused_variables)] extra_text_logger: Option<impl Log + 'static>,
227    ) -> Self {
228        STRUCTURED_LOG_BYTES.store(0, Ordering::Relaxed);
229
230        if let Some(state) = LOGGER_STATE.get() {
231            let mut active_sink = lock_mutex(&state.sink);
232            *active_sink = Box::new(sink);
233        } else {
234            let state = LoggerState {
235                sink: Mutex::new(Box::new(sink)),
236                clock,
237                live_listeners: Mutex::new(LiveLogListeners::default()),
238            };
239            init_logger_state(state);
240        }
241
242        // If caller provided a default text logger (std + debug builds), install it as the live listener.
243        #[cfg(all(feature = "std", debug_assertions))]
244        if let Some(logger) = extra_text_logger {
245            register_live_log_listener(move |entry, format_str, param_names| {
246                // Build a text line from structured data—no parsing.
247                let params: Vec<String> = entry.params.iter().map(|v| v.to_string()).collect();
248                let mut named_params = HashMap::new();
249                let mut param_names_iter = param_names.iter();
250                for (name_index, value) in entry.paramname_indexes.iter().zip(params.iter()) {
251                    if *name_index != cu29_log::ANONYMOUS {
252                        let Some(name) = param_names_iter.next() else {
253                            continue;
254                        };
255                        named_params.insert(name.to_string(), value.clone());
256                    }
257                }
258                if let Ok(line) = format_message_only(format_str, params.as_slice(), &named_params)
259                {
260                    logger.log(
261                        &log::Record::builder()
262                            .args(format_args!("{line}"))
263                            .level(match entry.level {
264                                CuLogLevel::Debug => log::Level::Debug,
265                                CuLogLevel::Info => log::Level::Info,
266                                CuLogLevel::Warning => log::Level::Warn,
267                                CuLogLevel::Error => log::Level::Error,
268                                CuLogLevel::Critical => log::Level::Error,
269                            })
270                            .target("cu29_log")
271                            .module_path_static(Some("cu29_log"))
272                            .file_static(Some("cu29_log"))
273                            .line(Some(0))
274                            .build(),
275                    );
276                }
277            });
278        }
279
280        LoggerRuntime {}
281    }
282
283    /// Flush the active structured-log sink, reporting storage failures.
284    #[doc(hidden)]
285    pub fn try_flush(&self) -> CuResult<()> {
286        if let Some(state) = LOGGER_STATE.get() {
287            lock_mutex(&state.sink).flush()?;
288        }
289        Ok(())
290    }
291
292    pub fn flush(&self) {
293        // no op in no_std TODO(gbin): check if it will be needed in no_std at some point.
294        if let Some(state) = LOGGER_STATE.get() {
295            let mut sink = lock_mutex(&state.sink);
296            let _ = sink.flush(); // ignore errors in no_std
297        } else {
298            #[cfg(feature = "std")]
299            eprintln!("cu29_log: Logger not initialized.");
300        }
301    }
302}
303
304impl Drop for LoggerRuntime {
305    fn drop(&mut self) {
306        self.flush();
307        // Assume on no-std that there is no buffering. TODO(gbin): check if this hold true.
308        if let Some(state) = LOGGER_STATE.get() {
309            let mut sink = lock_mutex(&state.sink);
310            *sink = Box::new(DummyWriteStream);
311        }
312    }
313}
314
315/// Function called from generated code to log data.
316/// It moves entry by design, it will be absorbed in the queue.
317#[inline(always)]
318fn log_inner(
319    entry: &mut CuLogEntry,
320    notify: bool,
321    format_str: &str,
322    param_names: &[&str],
323) -> CuResult<()> {
324    let Some(state) = LOGGER_STATE.get() else {
325        return Err("Logger not initialized.".into());
326    };
327    entry.time = state.clock.now();
328
329    let mut guard = lock_mutex(&state.sink);
330    guard.log(entry)?;
331    if let Some(bytes) = guard.last_log_bytes() {
332        STRUCTURED_LOG_BYTES.fetch_add(bytes, Ordering::Relaxed);
333    }
334
335    // Basic notification; richer context added in log_debug_mode.
336    if notify {
337        notify_live_listeners(entry, format_str, param_names);
338    }
339    Ok(())
340}
341
342/// Public entry point used in release / no-debug paths.
343#[inline(always)]
344pub fn log(entry: &mut CuLogEntry) -> CuResult<()> {
345    log_inner(entry, true, "", &[])
346}
347
348/// Returns the total number of bytes written to the structured log stream.
349pub fn structured_log_bytes_total() -> u64 {
350    STRUCTURED_LOG_BYTES.load(Ordering::Relaxed) as u64
351}
352
353/// This version of log is only compiled in debug mode
354/// This allows a normal logging framework to be bridged.
355#[cfg(debug_assertions)]
356pub fn log_debug_mode(
357    entry: &mut CuLogEntry,
358    _format_str: &str, // this is the missing info at runtime.
359    _param_names: &[&str],
360) -> CuResult<()> {
361    // Write structured log but avoid double-notifying live listeners here.
362    log_inner(entry, false, "", &[])?;
363
364    // and the bridging is only available in std.
365    #[cfg(feature = "std")]
366    extra_log(entry, _format_str, _param_names)?;
367
368    Ok(())
369}
370
371#[cfg(debug_assertions)]
372#[cfg(feature = "std")]
373fn extra_log(entry: &mut CuLogEntry, format_str: &str, param_names: &[&str]) -> CuResult<()> {
374    // Legacy text logging now goes through the live listener; keep this as a thin shim.
375    notify_live_listeners(entry, format_str, param_names);
376
377    Ok(())
378}
379
380/// Register a live log listener; subsequent logs invoke `cb`.
381pub fn register_live_log_listener<F>(cb: F) -> Option<LiveLogListenerId>
382where
383    F: Fn(&CuLogEntry, &str, &[&str]) + Send + Sync + 'static,
384{
385    LOGGER_STATE.get().map(|state| {
386        let mut guard = lock_mutex(&state.live_listeners);
387        guard.insert(Box::new(cb))
388    })
389}
390
391/// Register a scoped live log listener and remove it when the returned guard is dropped.
392pub fn scoped_live_log_listener<F>(cb: F) -> LiveLogListenerGuard
393where
394    F: Fn(&CuLogEntry, &str, &[&str]) + Send + Sync + 'static,
395{
396    LiveLogListenerGuard {
397        id: register_live_log_listener(cb),
398    }
399}
400
401/// Register automatic console output, used only when no monitor log pane is active.
402#[doc(hidden)]
403#[cfg(all(feature = "std", debug_assertions))]
404pub fn scoped_live_log_stdout_listener<F>(cb: F) -> LiveLogListenerGuard
405where
406    F: Fn(&CuLogEntry, &str, &[&str]) + Send + Sync + 'static,
407{
408    scoped_live_log_output_listener(cb, LiveLogOutput::Stdout)
409}
410
411/// Register a monitor log pane and suspend automatic console output until it closes.
412#[doc(hidden)]
413#[cfg(all(feature = "std", debug_assertions))]
414pub fn scoped_live_log_monitor_listener<F>(cb: F) -> LiveLogListenerGuard
415where
416    F: Fn(&CuLogEntry, &str, &[&str]) + Send + Sync + 'static,
417{
418    scoped_live_log_output_listener(cb, LiveLogOutput::Monitor)
419}
420
421#[cfg(all(feature = "std", debug_assertions))]
422fn scoped_live_log_output_listener<F>(cb: F, output: LiveLogOutput) -> LiveLogListenerGuard
423where
424    F: Fn(&CuLogEntry, &str, &[&str]) + Send + Sync + 'static,
425{
426    LiveLogListenerGuard {
427        id: LOGGER_STATE.get().map(|state| {
428            let mut guard = lock_mutex(&state.live_listeners);
429            let id = guard.insert(Box::new(cb));
430            if let Some(listener) = guard.listeners.last_mut() {
431                listener.output = output;
432            }
433            id
434        }),
435    }
436}
437
438/// Remove a live log listener by id. No-op if runtime not initialized.
439pub fn unregister_live_log_listener_id(id: LiveLogListenerId) {
440    if let Some(state) = LOGGER_STATE.get() {
441        let mut guard = lock_mutex(&state.live_listeners);
442        guard.remove(id);
443    }
444}
445
446/// Remove all registered live log listeners. No-op if runtime not initialized.
447pub fn unregister_live_log_listener() {
448    if let Some(state) = LOGGER_STATE.get() {
449        let mut guard = lock_mutex(&state.live_listeners);
450        guard.listeners.clear();
451    }
452}
453
454/// Capture structured log entries emitted while `f` runs.
455///
456/// This is a scoped tee: existing live listeners still receive entries while
457/// capture is active.
458#[cfg(feature = "std")]
459pub fn capture_live_logs<F, R>(f: F) -> (R, Vec<CuLogEntry>)
460where
461    F: FnOnce() -> R,
462{
463    let captured_logs = std::sync::Arc::new(std::sync::Mutex::new(Vec::<CuLogEntry>::new()));
464    let listener_guard = LOGGER_STATE.get().map(|_| {
465        let captured_logs_for_listener = captured_logs.clone();
466        scoped_live_log_listener(move |entry, _, _| {
467            captured_logs_for_listener
468                .lock()
469                .expect("live log capture poisoned")
470                .push(entry.clone());
471        })
472    });
473
474    let result = f();
475    drop(listener_guard);
476    let logs = captured_logs
477        .lock()
478        .expect("live log capture poisoned")
479        .clone();
480
481    (result, logs)
482}
483
484/// Notify registered listener if any.
485#[allow(clippy::collapsible_if)]
486pub(crate) fn notify_live_listeners(entry: &CuLogEntry, format_str: &str, param_names: &[&str]) {
487    if let Some(state) = LOGGER_STATE.get() {
488        let guard = lock_mutex(&state.live_listeners);
489        guard.notify(entry, format_str, param_names);
490    }
491}
492// This is an adaptation of the Iowriter from bincode.
493
494#[cfg(feature = "std")]
495pub struct OwningIoWriter<W: Write> {
496    writer: BufWriter<W>,
497    bytes_written: usize,
498}
499
500#[cfg(feature = "std")]
501impl<W: Write> OwningIoWriter<W> {
502    pub fn new(writer: W) -> Self {
503        Self {
504            writer: BufWriter::new(writer),
505            bytes_written: 0,
506        }
507    }
508
509    pub fn bytes_written(&self) -> usize {
510        self.bytes_written
511    }
512
513    pub fn flush(&mut self) -> Result<(), EncodeError> {
514        self.writer.flush().map_err(|inner| EncodeError::Io {
515            inner,
516            index: self.bytes_written,
517        })
518    }
519}
520
521#[cfg(feature = "std")]
522impl<W: Write> Writer for OwningIoWriter<W> {
523    #[inline(always)]
524    fn write(&mut self, bytes: &[u8]) -> Result<(), EncodeError> {
525        self.writer
526            .write_all(bytes)
527            .map_err(|inner| EncodeError::Io {
528                inner,
529                index: self.bytes_written,
530            })?;
531        self.bytes_written += bytes.len();
532        Ok(())
533    }
534}
535
536/// This allows this crate to be used outside of Copper (ie. decoupling it from the unifiedlog.
537#[cfg(feature = "std")]
538pub struct SimpleFileWriter {
539    path: PathBuf,
540    encoder: EncoderImpl<OwningIoWriter<File>, Configuration>,
541}
542
543#[cfg(feature = "std")]
544impl SimpleFileWriter {
545    pub fn new(path: &PathBuf) -> CuResult<Self> {
546        let file = std::fs::OpenOptions::new()
547            .create(true)
548            .truncate(true)
549            .write(true)
550            .open(path)
551            .map_err(|e| format!("Failed to open file: {e:?}"))?;
552
553        let writer = OwningIoWriter::new(file);
554        let encoder = EncoderImpl::new(writer, bincode::config::standard());
555
556        Ok(SimpleFileWriter {
557            path: path.clone(),
558            encoder,
559        })
560    }
561}
562
563#[cfg(feature = "std")]
564impl Debug for SimpleFileWriter {
565    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
566        write!(f, "SimpleFileWriter for path {:?}", self.path)
567    }
568}
569
570#[cfg(feature = "std")]
571impl WriteStream<CuLogEntry> for SimpleFileWriter {
572    #[inline(always)]
573    fn log(&mut self, obj: &CuLogEntry) -> CuResult<()> {
574        obj.encode(&mut self.encoder)
575            .map_err(|e| format!("Failed to write to file: {e:?}"))?;
576        Ok(())
577    }
578
579    fn flush(&mut self) -> CuResult<()> {
580        self.encoder
581            .writer()
582            .flush()
583            .map_err(|e| format!("Failed to flush file: {e:?}"))?;
584        Ok(())
585    }
586}
587
588#[cfg(test)]
589mod tests {
590    use crate::CuLogEntry;
591    use bincode::config::standard;
592    use cu29_log::{CuLogLevel, CuLogOrigin};
593    use cu29_value::Value;
594    use smallvec::smallvec;
595
596    #[cfg(not(feature = "std"))]
597    use alloc::string::ToString;
598
599    #[cfg(all(feature = "std", debug_assertions))]
600    #[test]
601    fn test_monitor_routes_logs_without_stdout_and_preserves_observers() {
602        use crate::{LiveLogListeners, LiveLogOutput};
603        use std::sync::{Arc, Mutex};
604
605        let calls = Arc::new(Mutex::new(Vec::new()));
606        let mut listeners = LiveLogListeners::default();
607        let mut add = |name, output| {
608            let calls = calls.clone();
609            let id = listeners.insert(Box::new(move |_, _, _| {
610                calls.lock().unwrap().push(name);
611            }));
612            listeners.listeners.last_mut().unwrap().output = output;
613            id
614        };
615        let stdout = add("stdout", LiveLogOutput::Stdout);
616        add("other stdout", LiveLogOutput::Stdout);
617        add("observer", LiveLogOutput::Observer);
618        let monitor = add("monitor", LiveLogOutput::Monitor);
619        let entry = CuLogEntry {
620            time: 0.into(),
621            level: CuLogLevel::Debug,
622            origin: CuLogOrigin::default(),
623            msg_index: 1,
624            paramname_indexes: smallvec![],
625            params: smallvec![],
626        };
627
628        // Both runtimes' automatic stdout listeners are suppressed while the pane is active.
629        listeners.notify(&entry, "rc channels", &[]);
630        assert_eq!(*calls.lock().unwrap(), ["observer", "monitor"]);
631
632        // Closing the pane resumes headless output without removing the independent observer.
633        listeners.remove(monitor);
634        calls.lock().unwrap().clear();
635        listeners.notify(&entry, "rc channels", &[]);
636        assert_eq!(*calls.lock().unwrap(), ["stdout", "observer"]);
637
638        // Removing one runtime's listener leaves the other runtime's fallback available.
639        listeners.remove(stdout);
640        calls.lock().unwrap().clear();
641        listeners.notify(&entry, "rc channels", &[]);
642        assert_eq!(*calls.lock().unwrap(), ["other stdout", "observer"]);
643    }
644
645    #[test]
646    fn test_encode_decode_structured_log() {
647        let log_entry = CuLogEntry {
648            time: 0.into(),
649            level: CuLogLevel::Info,
650            origin: CuLogOrigin::default(),
651            msg_index: 1,
652            paramname_indexes: smallvec![2, 3],
653            params: smallvec![Value::String("test".to_string())],
654        };
655        let encoded = bincode::encode_to_vec(&log_entry, standard()).unwrap();
656        let decoded_tuple: (CuLogEntry, usize) =
657            bincode::decode_from_slice(&encoded, standard()).unwrap();
658        assert_eq!(log_entry, decoded_tuple.0);
659    }
660
661    #[cfg(all(feature = "std", debug_assertions))]
662    #[test]
663    fn test_format_message_only_mixes_named_and_positional_placeholders() {
664        let params = vec!["event payload".to_string()];
665        let mut named_params = std::collections::HashMap::new();
666        named_params.insert("hash".to_string(), "0x000000000".to_string());
667        named_params.insert("size".to_string(), "420".to_string());
668
669        let formatted = crate::format_message_only(
670            "File closed after hash was calculated Hash: {hash}, size: {size};\n{}",
671            &params,
672            &named_params,
673        )
674        .unwrap();
675
676        assert_eq!(
677            formatted,
678            "File closed after hash was calculated Hash: 0x000000000, size: 420;\nevent payload"
679        );
680    }
681}
cu29_log_runtime/lib.rs

cu29_log_runtime/
lib.rs
