fix(diagnostics): bound formatting and release ingress before callbacks

Bound UTF-8 formatting and release diagnostic ingress before invoking
callbacks. Check error-ID collisions at Rust/C++ release final link.

Validation: diagnostic tests, strict Clippy, Miri and final-link cases.

firmware/libs/kaptan-log/src/facade.rs

@@ -71,18 +71,18 @@ impl<const N: usize> LogIngress<N> {
     }
 
     fn drain_one<R>(&self, consume: impl FnOnce(&LogRecord) -> R) -> Option<R> {
-        self.with_state(|entries, head, len| {
-            if *len == 0 {
-                return None;
-            }
-            let record = entries[*head].as_ref()?;
-            let result = consume(record);
-            entries[*head] = None;
-            *head = (*head + 1) % N;
-            *len -= 1;
-            Some(result)
-        })
-        .flatten()
+        let record = self
+            .with_state(|entries, head, len| {
+                if *len == 0 {
+                    return None;
+                }
+                let record = entries[*head].take();
+                *head = (*head + 1) % N;
+                *len -= 1;
+                record
+            })
+            .flatten()?;
+        Some(consume(&record))
     }
 
     fn with_state<R>(
@@ -116,8 +116,9 @@ pub fn publish(level: Level, arguments: fmt::Arguments<'_>) {
 
 /// Visits and removes the oldest pending log event.
 ///
-/// The callback runs during the ingress's one nonblocking ownership attempt;
-/// it should perform only bounded composition-level sink fanout.
+/// The record is removed and exclusive ingress access is released before the
+/// callback runs. Reentrant logging can use the freed slot. If the callback
+/// unwinds, this record is consumed and other pending records remain available.
 #[inline]
 pub fn drain_one<R>(consume: impl FnOnce(&LogRecord) -> R) -> Option<R> {
     GLOBAL_INGRESS.drain_one(consume)
@@ -138,6 +139,70 @@ mod tests {
         core::str::from_utf8(&text.bytes()[..usize::from(text.byte_length())]).unwrap()
     }
 
+    #[test]
+    fn concurrent_publication_and_drain_preserve_complete_unique_records() {
+        let ingress = LogIngress::<2>::new();
+        // A critical record cannot be evicted. This guarantees one observation
+        // even if contention rejects every concurrent publication.
+        ingress.publish(Level::Error, core::format_args!("initial"));
+        let start = std::sync::Barrier::new(3);
+        let mut records = std::thread::scope(|scope| {
+            let ingress = &ingress;
+            let start = &start;
+            let producers: [_; 2] = core::array::from_fn(|producer| {
+                scope.spawn(move || {
+                    start.wait();
+                    for sequence in 0..8 {
+                        ingress.publish(
+                            [Level::Info, Level::Error][producer],
+                            core::format_args!("{producer}:{sequence}"),
+                        );
+                        std::thread::yield_now();
+                    }
+                })
+            });
+            let consumer = scope.spawn(move || {
+                start.wait();
+                let mut records = std::vec::Vec::new();
+                for _ in 0..64 {
+                    records.extend(ingress.drain_one(|record| *record));
+                    std::thread::yield_now();
+                }
+                records
+            });
+            for producer in producers {
+                producer.join().unwrap();
+            }
+            consumer.join().unwrap()
+        });
+        for _ in 0..2 {
+            records.extend(ingress.drain_one(|record| *record));
+        }
+        assert!(ingress.drain_one(|_| ()).is_none());
+
+        // Contention and critical replacement may discard publications. A
+        // delivered record must retain its own level and text without repeats.
+        let mut seen = [false; 16];
+        let mut initial_seen = false;
+        for record in records {
+            if message(&record) == "initial" {
+                assert_eq!(record.level(), Level::Error);
+                assert!(!core::mem::replace(&mut initial_seen, true));
+                continue;
+            }
+            let (producer, sequence) = message(&record).split_once(':').unwrap();
+            let producer: usize = producer.parse().unwrap();
+            let sequence: usize = sequence.parse().unwrap();
+            assert!(producer < 2 && sequence < 8);
+            assert_eq!(record.level(), [Level::Info, Level::Error][producer]);
+            assert!(!core::mem::replace(
+                &mut seen[producer * 8 + sequence],
+                true
+            ));
+        }
+        assert!(initial_seen);
+    }
+
     #[test]
     fn every_macro_produces_the_same_log_record_kind() {
         let _guard = GLOBAL_INGRESS_TEST
@@ -213,4 +278,48 @@ mod tests {
             Some(())
         );
     }
+
+    #[test]
+    fn callback_can_publish_and_drain_reentrantly() {
+        let ingress = LogIngress::<1>::new();
+        ingress.publish(Level::Info, core::format_args!("outer"));
+        assert_eq!(
+            ingress.drain_one(|record| {
+                assert!(!ingress.busy.load(Ordering::Relaxed));
+                ingress.publish(Level::Info, core::format_args!("inner"));
+                assert_eq!(
+                    ingress.drain_one(|inner| assert_eq!(message(inner), "inner")),
+                    Some(())
+                );
+                assert_eq!(message(record), "outer");
+            }),
+            Some(())
+        );
+        assert!(ingress.drain_one(|_| ()).is_none());
+    }
+
+    #[test]
+    fn callback_unwind_leaves_ingress_free_and_other_records_pending() {
+        let ingress = LogIngress::<2>::new();
+        ingress.publish(Level::Info, core::format_args!("consumed"));
+        ingress.publish(Level::Error, core::format_args!("pending"));
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            ingress.drain_one(|record| {
+                assert_eq!(message(record), "consumed");
+                panic!("consumer failed");
+            });
+        }));
+        assert!(result.is_err());
+        assert!(!ingress.busy.load(Ordering::Relaxed));
+        ingress.publish(Level::Info, core::format_args!("new"));
+        assert_eq!(
+            ingress.drain_one(|record| assert_eq!(message(record), "pending")),
+            Some(())
+        );
+        assert_eq!(
+            ingress.drain_one(|record| assert_eq!(message(record), "new")),
+            Some(())
+        );
+        assert!(ingress.drain_one(|_| ()).is_none());
+    }
 }

firmware/libs/kaptan-log/src/lib.rs

@@ -1,12 +1,11 @@
 //! Bounded, transport-neutral runtime logging for Kaptan.
 //!
-//! This crate deliberately contains no MAVLink, board, filesystem, or clock
-//! dependency. Its sole configuration coupling is compile-time enum metadata
-//! for [`LevelFilter`]. Flight code reports bounded events to
-//! independently filtered sink queues; a platform-owned adapter may later
-//! encode one queue for a link.
-//! Reporting is observational and must never influence flight control or
-//! actuator authority.
+//! Each sink filters and queues events for a transport adapter. Logging must
+//! not affect flight-control decisions or actuator authority.
+//!
+//! Text overflow stops formatting that propagates `core::fmt::Error`. Custom
+//! `Display` implementations may ignore that error or perform arbitrary work,
+//! so fixed storage and queue bounds do not establish a universal time bound.
 
 #![no_std]
 

firmware/libs/kaptan-log/src/text.rs

@@ -27,6 +27,9 @@ impl LogText {
     }
 
     /// Formats one message without allocation.
+    ///
+    /// Overflow stops formatters that propagate `fmt::Error`. A custom `Display`
+    /// implementation can ignore that error or do unbounded work of its own.
     pub(crate) fn from_arguments(arguments: fmt::Arguments<'_>) -> Self {
         let mut text = Self::new();
         let _ = fmt::write(&mut text, arguments);
@@ -58,6 +61,9 @@ impl LogText {
 
 impl fmt::Write for LogText {
     fn write_str(&mut self, message: &str) -> fmt::Result {
+        if self.is_truncated() {
+            return Err(fmt::Error);
+        }
         let remaining = MAXIMUM_LOG_TEXT_BYTES - usize::from(self.length);
         let prefix = if message.len() <= remaining {
             message.len()
@@ -73,13 +79,18 @@ impl fmt::Write for LogText {
         let used = usize::from(self.length);
         self.bytes[used..used + prefix].copy_from_slice(&message.as_bytes()[..prefix]);
         self.length = (used + prefix) as u8;
-        Ok(())
+        if self.is_truncated() {
+            Err(fmt::Error)
+        } else {
+            Ok(())
+        }
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::{LogText, MAXIMUM_LOG_TEXT_BYTES};
+    use core::fmt::{self, Write};
 
     #[test]
     fn truncates_only_at_utf8_boundaries() {
@@ -95,4 +106,61 @@ mod tests {
             MAXIMUM_LOG_TEXT_BYTES - 1
         );
     }
+
+    #[test]
+    fn exact_multibyte_fit_is_not_truncated() {
+        let input = "x".repeat(MAXIMUM_LOG_TEXT_BYTES - 2) + "é";
+        let mut text = LogText::new();
+        assert_eq!(text.write_str(&input), Ok(()));
+        assert_eq!(text.write_str(""), Ok(()));
+        assert!(!text.is_truncated());
+        assert_eq!(text.bytes().as_slice(), input.as_bytes());
+        assert_eq!(text.write_str("a"), Err(fmt::Error));
+        assert!(text.is_truncated());
+        assert_eq!(text.bytes().as_slice(), input.as_bytes());
+    }
+
+    #[test]
+    fn overflow_latches_even_if_a_formatter_ignores_the_error() {
+        struct Fragments;
+        impl fmt::Display for Fragments {
+            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str(&"x".repeat(MAXIMUM_LOG_TEXT_BYTES - 3))?;
+                formatter.write_str("é")?;
+                let _ = formatter.write_str("🙂");
+                let _ = formatter.write_str("a");
+                Ok(())
+            }
+        }
+        let text = LogText::from_arguments(format_args!("{Fragments}"));
+        let expected = "x".repeat(MAXIMUM_LOG_TEXT_BYTES - 3) + "é";
+        assert!(text.is_truncated());
+        assert_eq!(
+            &text.bytes()[..usize::from(text.byte_length())],
+            expected.as_bytes()
+        );
+        assert_eq!(text.bytes()[MAXIMUM_LOG_TEXT_BYTES - 1], 0);
+    }
+
+    #[test]
+    fn large_standard_padding_stops_at_the_first_overflow() {
+        struct CountWrites {
+            text: LogText,
+            writes: usize,
+        }
+        impl fmt::Write for CountWrites {
+            fn write_str(&mut self, message: &str) -> fmt::Result {
+                self.writes += 1;
+                self.text.write_str(message)
+            }
+        }
+        let mut output = CountWrites {
+            text: LogText::new(),
+            writes: 0,
+        };
+        assert_eq!(write!(output, "{:>65535}", "end"), Err(fmt::Error));
+        assert!(output.writes <= MAXIMUM_LOG_TEXT_BYTES + 1);
+        assert!(output.text.is_truncated());
+        assert_eq!(output.text.bytes(), &[b' '; MAXIMUM_LOG_TEXT_BYTES]);
+    }
 }