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

//! Bounded priority handling for fixed human-readable diagnostic records.

use core::fmt;

use crate::{
    DeliveryClass, ErrorAdditionalInformation, ErrorPath, ErrorReport, LogFilter, LogFlags,
    LogRecord, LogText, Severity, StaticError, StaticErrorMailbox,
};

/// Outcome from one bounded diagnostic admission attempt.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PublishOutcome {
    /// The record was retained.
    Queued,
    /// A record was excluded by this sink's configured policy.
    Filtered,
    /// A best-effort record was replaced by newer optional information.
    ReplacedOldest,
    /// The critical lane kept its earlier record.
    RejectedFull,
}

/// Result returned by the sole telemetry owner after attempting to admit a
/// record to its lower transport.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransportAdmission {
    /// The lower transport retained the record, so it may leave the queue.
    Accepted,
    /// The lower transport did not accept it yet; retain it for retry.
    Retry,
}

struct Lane<const N: usize> {
    entries: [Option<LogRecord>; N],
    head: usize,
    len: usize,
}

impl<const N: usize> Lane<N> {
    const fn new() -> Self {
        assert!(N > 0, "diagnostic lanes must be nonempty");
        Self {
            entries: [None; N],
            head: 0,
            len: 0,
        }
    }

    fn publish(&mut self, record: LogRecord, replace_oldest: bool) -> PublishOutcome {
        if self.len == N {
            if replace_oldest {
                self.entries[self.head] = Some(record);
                self.head = (self.head + 1) % N;
                return PublishOutcome::ReplacedOldest;
            }
            return PublishOutcome::RejectedFull;
        }
        let index = (self.head + self.len) % N;
        self.entries[index] = Some(record);
        self.len += 1;
        PublishOutcome::Queued
    }

    const fn is_full(&self) -> bool {
        self.len == N
    }

    fn peek(&self) -> Option<LogRecord> {
        self.entries.get(self.head).copied().flatten()
    }

    fn pop(&mut self) -> Option<LogRecord> {
        if self.len == 0 {
            return None;
        }
        let record = self.entries[self.head].take();
        self.head = (self.head + 1) % N;
        self.len -= 1;
        record
    }
}

/// One bounded priority queue owned by a log sink.
///
/// It owns no global logger, MAVLink state, DMA, filesystem, or other sink.
pub struct LogSinkQueue<const CRITICAL: usize, const BEST_EFFORT: usize> {
    filter: LogFilter,
    critical: Lane<CRITICAL>,
    best_effort: Lane<BEST_EFFORT>,
}

impl<const CRITICAL: usize, const BEST_EFFORT: usize> LogSinkQueue<CRITICAL, BEST_EFFORT> {
    /// Creates one sink queue with its immutable boot-active policy.
    pub const fn new(filter: LogFilter) -> Self {
        Self {
            filter,
            critical: Lane::new(),
            best_effort: Lane::new(),
        }
    }

    /// Immutable sink policy.
    pub const fn filter(&self) -> LogFilter {
        self.filter
    }

    /// Whether an error source may be consumed without losing an enabled
    /// critical record. Filtered errors may be consumed intentionally.
    pub const fn can_publish_error(&self) -> bool {
        !self.filter.allows(Severity::Error) || !self.critical.is_full()
    }

    /// Consumes at most one source-local static-error mailbox in round-robin
    /// order. An enabled source remains untouched while the critical lane is
    /// full; a filtered source is consumed intentionally.
    pub fn drain_next_error_mailbox<const SOURCES: usize>(
        &mut self,
        mailboxes: [&StaticErrorMailbox; SOURCES],
        cursor: &mut u8,
    ) -> bool {
        assert!(
            SOURCES != 0 && SOURCES <= u8::MAX as usize,
            "error mailbox fan-in must fit its u8 cursor"
        );
        if !self.can_publish_error() {
            return false;
        }
        if usize::from(*cursor) >= SOURCES {
            *cursor = 0;
        }

        let mut checked = 0;
        while checked < SOURCES {
            let index = usize::from(*cursor);
            *cursor = if index + 1 == SOURCES { 0 } else { *cursor + 1 };
            checked += 1;

            let (error, coalesced) = mailboxes[index].take();
            let Some(error) = error else {
                continue;
            };
            let flags = if coalesced {
                LogFlags::COALESCED
            } else {
                LogFlags::NONE
            };
            let outcome = self.publish_error_static(flags, error);
            debug_assert!(matches!(
                outcome,
                PublishOutcome::Queued | PublishOutcome::Filtered
            ));
            return true;
        }
        false
    }

    /// Formats and retains one ordinary task/worker diagnostic. Error-level
    /// severities are emitted through `KAPTAN_ERROR`; lower severities use
    /// `KAPTAN_LOG`.
    pub fn publish_arguments(
        &mut self,
        severity: Severity,
        flags: LogFlags,
        arguments: fmt::Arguments<'_>,
    ) -> PublishOutcome {
        self.publish_text(severity, flags, LogText::from_arguments(arguments))
    }

    /// Retains one static derived error without formatting or exposing source
    /// values. The adapter maps it to `KAPTAN_ERROR`.
    pub fn publish_error<E: ErrorReport>(&mut self, flags: LogFlags, error: &E) -> PublishOutcome {
        self.publish_record(LogRecord::from_error(flags, error))
    }

    /// Retains one static mailbox error. This is the ISR-safe counterpart to
    /// [`Self::publish_error`].
    pub fn publish_error_static(
        &mut self,
        flags: LogFlags,
        error: &'static StaticError,
    ) -> PublishOutcome {
        self.publish_record(LogRecord::from_static_error(flags, error))
    }

    /// Retains a preformatted error identity and optional detail. Both
    /// transport-neutral components already have the exact common diagnostic
    /// envelope capacities, so a sink adapter need not retain or clip them
    /// again.
    pub fn publish_error_text(
        &mut self,
        flags: LogFlags,
        path: ErrorPath,
        additional_information: ErrorAdditionalInformation,
    ) -> PublishOutcome {
        self.publish_record(LogRecord::error(flags, path, additional_information))
    }

    /// Retains one preformatted bounded text payload. Error-or-higher
    /// severities are independently encoded as `KAPTAN_ERROR`.
    pub fn publish_text(
        &mut self,
        severity: Severity,
        flags: LogFlags,
        text: LogText,
    ) -> PublishOutcome {
        self.publish_record(LogRecord::new(severity, flags, text))
    }

    /// Applies this sink's policy and retains one already-constructed event.
    ///
    /// Composition may submit the same transport-neutral record to multiple
    /// sink queues without coupling event creation to any sink policy.
    pub fn publish_record(&mut self, record: LogRecord) -> PublishOutcome {
        if !self.filter.allows(record.severity()) {
            return PublishOutcome::Filtered;
        }
        let delivery = record.delivery();
        match delivery {
            DeliveryClass::Critical => self.critical.publish(record, false),
            DeliveryClass::BestEffort => self.best_effort.publish(record, true),
        }
    }

    /// Highest-priority record without relinquishing it.
    pub fn peek(&self) -> Option<LogRecord> {
        self.critical.peek().or_else(|| self.best_effort.peek())
    }

    /// Submits the current record through the sink owner and removes it only
    /// after `Accepted`. This keeps encoder/admission retry semantics local
    /// for every severity.
    pub fn submit_next<R>(
        &mut self,
        submit: impl FnOnce(&LogRecord) -> (TransportAdmission, R),
    ) -> Option<R> {
        let record = self.peek()?;
        let (admission, result) = submit(&record);
        match admission {
            TransportAdmission::Accepted => {
                let _ = self.pop();
            }
            TransportAdmission::Retry => {}
        }
        Some(result)
    }

    fn pop(&mut self) -> Option<LogRecord> {
        self.critical.pop().or_else(|| self.best_effort.pop())
    }
}

#[cfg(test)]
mod tests {
    use super::{LogSinkQueue, PublishOutcome, TransportAdmission};
    use crate::{
        DeliveryClass, ErrorAdditionalInformation, ErrorPath, LogFilter, LogFlags, LogRecord,
        LogText, RecordChannel, Severity, StaticError,
    };

    #[test]
    fn sink_policy_applies_to_errors_and_ordinary_logs() {
        let mut disabled = LogSinkQueue::<1, 1>::new(LogFilter::disabled());
        assert_eq!(
            disabled.publish_text(
                Severity::Error,
                LogFlags::NONE,
                LogText::from_message("error")
            ),
            PublishOutcome::Filtered
        );
        assert!(disabled.peek().is_none());

        let mut queue = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Info));
        assert_eq!(
            queue.publish_text(
                Severity::Debug,
                LogFlags::NONE,
                LogText::from_message("debug")
            ),
            PublishOutcome::Filtered
        );
        assert_eq!(
            queue.publish_text(
                Severity::Error,
                LogFlags::NONE,
                LogText::from_message("error")
            ),
            PublishOutcome::Queued
        );
        let record = queue.peek().unwrap();
        assert_eq!(record.error_path().unwrap().as_str(), "error");
        assert!(record.additional_information().unwrap().is_empty());
        assert_eq!(record.channel(), RecordChannel::Error);
    }

    #[test]
    fn full_critical_lane_backpressures_only_an_enabled_error_source() {
        let mut enabled = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Info));
        assert!(enabled.can_publish_error());
        assert_eq!(
            enabled.publish_text(
                Severity::Error,
                LogFlags::NONE,
                LogText::from_message("pending")
            ),
            PublishOutcome::Queued
        );
        assert!(!enabled.can_publish_error());

        let disabled = LogSinkQueue::<1, 1>::new(LogFilter::disabled());
        assert!(disabled.can_publish_error());
    }

    #[test]
    fn mailbox_fan_in_survives_backpressure_and_is_round_robin() {
        static FIRST: StaticError = StaticError::new("first");
        static SECOND: StaticError = StaticError::new("second");
        static THIRD: StaticError = StaticError::new("third");
        let first = crate::StaticErrorMailbox::new();
        let second = crate::StaticErrorMailbox::new();
        let third = crate::StaticErrorMailbox::new();
        let mailboxes = [&first, &second, &third];
        first.record(&FIRST);
        second.record(&SECOND);
        third.record(&THIRD);

        let mut queue = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Info));
        let mut cursor = 0;
        assert!(queue.drain_next_error_mailbox(mailboxes, &mut cursor));
        assert_eq!(
            queue.peek().unwrap().error_path().unwrap().as_str(),
            "first"
        );

        first.record(&FIRST);
        assert!(!queue.drain_next_error_mailbox(mailboxes, &mut cursor));
        assert_eq!(
            queue.peek().unwrap().error_path().unwrap().as_str(),
            "first"
        );

        let mut expect = |path| {
            assert_eq!(queue.peek().unwrap().error_path().unwrap().as_str(), path);
            assert_eq!(
                queue.submit_next(|_| (TransportAdmission::Accepted, ())),
                Some(())
            );
            assert!(queue.drain_next_error_mailbox(mailboxes, &mut cursor));
        };
        expect("first");
        expect("second");
        expect("third");
        assert_eq!(
            queue.peek().unwrap().error_path().unwrap().as_str(),
            "first"
        );
    }

    #[test]
    fn identical_events_remain_distinct_records() {
        static FAULT: StaticError = StaticError::new("driver::Fault");
        let mut queue = LogSinkQueue::<2, 1>::new(LogFilter::up_to(Severity::Debug));
        assert_eq!(
            queue.publish_error_static(LogFlags::NONE, &FAULT),
            PublishOutcome::Queued
        );
        assert_eq!(
            queue.publish_error_static(LogFlags::NONE, &FAULT),
            PublishOutcome::Queued
        );
        let record = queue.peek().unwrap();
        assert!(!record.flags().contains(LogFlags::COALESCED));
        assert_eq!(record.error_path().unwrap().as_str(), "driver::Fault");
        assert!(record.additional_information().unwrap().is_empty());
        assert_eq!(
            queue.submit_next(|_| (TransportAdmission::Accepted, ())),
            Some(())
        );
        assert_eq!(queue.peek(), Some(record));
    }

    #[test]
    fn static_error_detail_needs_no_producer_formatting() {
        static FAULT: StaticError =
            StaticError::with_additional_information("driver::Dma::Fault", "stream 7");
        let mut hub = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Debug));
        assert_eq!(
            hub.publish_error_static(LogFlags::NONE, &FAULT),
            PublishOutcome::Queued
        );
        let record = hub.peek().unwrap();
        assert_eq!(record.error_path().unwrap().as_str(), "driver::Dma::Fault");
        assert_eq!(
            record.additional_information().unwrap().as_str(),
            "stream 7"
        );
    }

    #[test]
    fn error_identity_and_detail_are_independent_record_components() {
        let mut hub = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Debug));
        let _ = hub.publish_error_text(
            LogFlags::NONE,
            ErrorPath::from_message("driver::Icm42688::Transfer"),
            ErrorAdditionalInformation::from_message("SPI DMA timeout"),
        );
        let record = hub.peek().unwrap();
        assert_eq!(record.channel(), RecordChannel::Error);
        assert_eq!(
            record.error_path().unwrap().as_str(),
            "driver::Icm42688::Transfer"
        );
        assert_eq!(
            record.additional_information().unwrap().as_str(),
            "SPI DMA timeout"
        );
    }

    #[test]
    fn retry_retains_every_severity() {
        let mut hub = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Debug));
        let _ = hub.publish_text(
            Severity::Error,
            LogFlags::NONE,
            LogText::from_message("fault"),
        );
        assert_eq!(
            hub.submit_next(|_| (TransportAdmission::Retry, ())),
            Some(())
        );
        assert_eq!(hub.peek().unwrap().delivery(), DeliveryClass::Critical);
        assert_eq!(
            hub.submit_next(|_| (TransportAdmission::Accepted, ())),
            Some(())
        );
        assert!(hub.peek().is_none());

        let _ = hub.publish_text(
            Severity::Info,
            LogFlags::NONE,
            LogText::from_message("detail"),
        );
        assert_eq!(
            hub.submit_next(|_| (TransportAdmission::Retry, ())),
            Some(())
        );
        assert_eq!(hub.peek().unwrap().delivery(), DeliveryClass::BestEffort);
        assert_eq!(
            hub.submit_next(|_| (TransportAdmission::Accepted, ())),
            Some(())
        );
        assert!(hub.peek().is_none());
    }

    #[test]
    fn independent_sink_policies_do_not_suppress_each_other() {
        let record = LogRecord::new(
            Severity::Info,
            LogFlags::NONE,
            LogText::from_message("flight ready"),
        );
        let mut mavlink = LogSinkQueue::<1, 1>::new(LogFilter::disabled());
        let mut blackbox = LogSinkQueue::<1, 1>::new(LogFilter::up_to(Severity::Debug));

        assert_eq!(mavlink.publish_record(record), PublishOutcome::Filtered);
        assert_eq!(blackbox.publish_record(record), PublishOutcome::Queued);
        assert!(mavlink.peek().is_none());
        assert_eq!(blackbox.peek(), Some(record));
    }
}