nros_log/

lib.rs

//! Phase 88 — portable leveled-logging facade for nano-ros.
//!
//! See [`docs/roadmap/archived/phase-88-nros-log.md`](../../../docs/roadmap/archived/phase-88-nros-log.md)
//! for the design and acceptance criteria.
//!
//! ## Layering
//!
//! - This crate carries only the portable types + dispatcher +
//!   macros + `PlatformSink`. No backend code.
//! - Per-platform log delivery is the responsibility of each
//!   `nros-platform-<rtos>` crate, exposing
//!   `nros_platform_log_write` / `nros_platform_log_flush` via the
//!   `nros_platform_*` ABI (header at
//!   `packages/platform/nros-platform-api/include/nros/platform.h`).
//! - `PlatformSink` is the bridge: a single `LogSink` impl that
//!   forwards to the ABI. Apps that want fan-out (e.g.
//!   `Platform + /rosout`) compose a `&'static [&dyn LogSink]`
//!   manually and pass it to [`init`].
//!
//! ## Quick start
//!
//! ```ignore
//! use nros_log::{Logger, Severity};
//! use nros_log::{log_info, log_warn};
//!
//! static LOGGER: Logger = Logger::new("my_node");
//!
//! fn main() {
//!     nros_log::register_logger(&LOGGER);
//!     nros_log::init(nros_log::sinks::default());
//!     log_info!(&LOGGER, "started; domain = {}", 42);
//! }
//! ```

// UNCONDITIONAL. This was `cfg_attr(not(feature = "std"), no_std)`, which
// bought nothing: this crate's `std` feature is `std = ["alloc"]` over
// `alloc = []`, so it enables no std functionality and the crate uses none.
// The conditional form only meant that building with `std` compiled a
// DIFFERENT crate -- a second configuration nobody wanted and no lane checked.
//
// ARCHITECTURE section 2 / phase-359: the terminal state of a core crate is
// `core` or `core+alloc`. `check-core-crates-are-no-std` holds every core
// crate to this spelling.
#![no_std]
#![deny(unsafe_op_in_unsafe_fn)]
#![warn(missing_docs)]

#[cfg(feature = "alloc")]
extern crate alloc;

// Phase 88.16.E — portable-atomic polyfill for CAS-less targets
// (RISC-V `imc`, etc.). Feature unification: a consuming bare-metal
// crate enables `unsafe-assume-single-core` / `critical-section` on
// its own `portable-atomic` dep; native CAS targets get the
// passthrough.
use portable_atomic::{AtomicBool, AtomicPtr, AtomicU8, AtomicUsize, Ordering};

pub mod early;
#[cfg(feature = "log-compat")]
pub mod log_compat;
pub mod log_params;
pub mod macros;
pub mod pool;
#[cfg(feature = "rosout")]
pub mod rosout;
pub mod sinks;
pub mod throttle;

mod buffer;

/// The resolved `[knobs.log]` tenant, written by `build.rs` (phase-479 W5,
/// issue 1037). One file, one place each value enters the crate.
mod config {
    include!(concat!(env!("OUT_DIR"), "/nros_log_config.rs"));

    // `build.rs` range-checks the knob; this keeps a hand-edited or stale
    // file from compiling into a ceiling above `off`.
    #[allow(clippy::absurd_extreme_comparisons)]
    const _: () = assert!(MAX_LEVEL <= 6, "NROS_LOG_MAX_LEVEL is 0..=6");
}

pub use buffer::{FormatBuffer, format_buffer_capacity};
pub use log_params::{LogOccurrence, LogParams, ThrottleClock, ToLogParams};
pub use pool::{
    MAX_LOGGER_NAME_LEN, dynamic_logger_capacity, dynamic_logger_name_arena, dynamic_loggers_in_use,
};
pub use throttle::{ThrottleState, interval_ms_to_ns, throttle_decide};

/// REP-2012 severity levels: rcutils's five, in rcutils's order, plus `Trace`.
///
/// NOT `rcutils_log_severity_t`'s numbering (`UNSET=0, DEBUG=10 ... FATAL=50`),
/// and no `Unset` variant. The integer representation is stable and part of the
/// ABI for `nros_platform_log_write`, so every value must be a level a sink can
/// render; "no level of its own" is the verb [`Logger::unset_level`] instead.
/// (The C surface's `rcl_log_severity_t` does carry rcutils's numbers and
/// `UNSET`.) Lower value = more verbose.
#[repr(u8)]
#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub enum Severity {
    /// Per-instruction granularity. Off unless `max-level-trace` is
    /// the active ceiling.
    Trace = 0,
    /// Diagnostic information useful while developing.
    Debug = 1,
    /// Normal operation events worth surfacing once.
    Info = 2,
    /// Unexpected but recoverable conditions.
    Warn = 3,
    /// Errors the caller should surface; the system continues.
    Error = 4,
    /// Unrecoverable — the system is about to abort.
    Fatal = 5,
}

impl Severity {
    /// Short uppercase label suitable for log-line rendering.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Trace => "TRACE",
            Self::Debug => "DEBUG",
            Self::Info => "INFO",
            Self::Warn => "WARN",
            Self::Error => "ERROR",
            Self::Fatal => "FATAL",
        }
    }

    /// Stable `u8` discriminant for cross-ABI use.
    #[must_use]
    pub const fn as_u8(self) -> u8 {
        self as u8
    }

    /// This severity on **rcutils's number line** — the one every ROS 2
    /// artefact outside this crate uses.
    ///
    /// [`as_u8`](Self::as_u8) is this crate's own compact discriminant (`0..=5`,
    /// chosen so a level comparison is an integer compare). It is NOT what a
    /// `rcl_interfaces/msg/Log` carries, nor what `<rclc/log.h>`'s
    /// `rcl_log_severity_t` crosses the FFI as, nor what
    /// `rcutils_logging_set_logger_level` takes. Those are all rcutils's
    /// `DEBUG=10 / INFO=20 / WARN=30 / ERROR=40 / FATAL=50`.
    ///
    /// THE ONE SPELLING of that mapping (issue 1025's class: one number, two
    /// derivations, agreeing until the day they do not). `nros-c`'s
    /// `rcl_log_severity_t::from_facade` is built on this, and a test there
    /// pins the result against the values `<rclc/log.h>` declares — which is
    /// the strongest join available, since this crate cannot see that header.
    ///
    /// `Trace` takes **5**, a value inside rcutils's own `UNSET`..`DEBUG` gap:
    /// rcutils has no TRACE, and borrowing `DEBUG`'s number would make the two
    /// levels indistinguishable on the wire.
    #[must_use]
    pub const fn rcutils_level(self) -> u8 {
        match self {
            Self::Trace => 5,
            Self::Debug => 10,
            Self::Info => 20,
            Self::Warn => 30,
            Self::Error => 40,
            Self::Fatal => 50,
        }
    }
}

/// Reconstruct a [`Severity`] from its `u8` discriminant.
///
/// Returns `None` for `> 5`.
#[must_use]
pub const fn severity_from_u8(value: u8) -> Option<Severity> {
    match value {
        0 => Some(Severity::Trace),
        1 => Some(Severity::Debug),
        2 => Some(Severity::Info),
        3 => Some(Severity::Warn),
        4 => Some(Severity::Error),
        5 => Some(Severity::Fatal),
        _ => None,
    }
}

/// Compile-time ceiling check used by the `nros_*!` macros.
///
/// Returns `true` iff `severity` is at or above the image's
/// `NROS_LOG_MAX_LEVEL` (issue 1037; builtin `trace`, so everything). A
/// `const fn` over a build-script constant, so a macro call below the ceiling
/// folds to `if false { .. }` and its formatting is dead code — the elimination
/// the old `max-level-*` features gave, without a feature.
///
/// The ceiling is a BOUND, not a threshold: [`Logger::set_level`] still moves
/// each logger's runtime level, and a level below the ceiling is simply
/// unreachable ([`Logger::is_enabled`] answers `false` there). rcutils keeps
/// the same pair (`RCUTILS_LOG_MIN_SEVERITY_*` + `set_logger_level`).
#[must_use]
pub const fn severity_enabled_at_compile_time(severity: Severity) -> bool {
    // `config::MAX_LEVEL` is 6 for `off`, above `Fatal` (5): nothing passes.
    // The value is a build-script constant, and at its builtin (0 = trace) the
    // comparison is always true — that is the point, not a slip.
    #[allow(clippy::absurd_extreme_comparisons)]
    let enabled = (severity as u8) >= config::MAX_LEVEL;
    enabled
}

/// One log entry, handed to each [`LogSink`].
///
/// `message` is already formatted — sinks must NOT re-format.
#[derive(Debug)]
pub struct Record<'a> {
    /// Severity of the record.
    pub severity: Severity,
    /// Name of the originating [`Logger`].
    pub logger_name: &'a str,
    /// Formatted message text (no trailing newline).
    pub message: &'a str,
    /// File the macro invocation came from (`core::file!()`).
    pub file: &'static str,
    /// Line within `file` (`core::line!()`).
    pub line: u32,
    /// Monotonic timestamp in nanoseconds. `0` if unavailable.
    pub timestamp_ns: u64,
}

/// Backend a log record is delivered to.
///
/// Implementations must be `Sync` so the dispatcher can hold them
/// in `&'static [&dyn LogSink]`. ISR-safety is per-impl — see the
/// table in `docs/roadmap/archived/phase-88-nros-log.md`.
pub trait LogSink: Sync {
    /// Render `record`. Called only when the record's severity passes
    /// both the compile-time ceiling AND the [`Logger`]'s runtime
    /// threshold.
    fn log(&self, record: &Record<'_>);

    /// Optional flush hook (default no-op).
    fn flush(&self) {}
}

/// The stored byte meaning "this logger has NO level of its own", so its
/// threshold is whatever [`Logger::default_level`] currently says.
///
/// phase-467 (ledger rows `c:log_severity_t` + `rust:Logger::set_default_level`).
/// It cannot be `0`: that is `Severity::Trace`, the most verbose level, which
/// is why storing rcutils's `UNSET` in a level byte used to select TRACE here
/// where upstream restores a default. `u8::MAX` is outside the enum in the
/// other direction, and being ABOVE `Fatal` is load-bearing in
/// [`Logger::is_enabled`] — see the note there.
const LEVEL_UNSET: u8 = u8::MAX;

/// The threshold a logger with no level of its own resolves to.
///
/// ONE process-wide byte, which is the whole of our level hierarchy. rcutils
/// resolves an unset logger by walking a DOTTED ANCESTRY (`x.y.z` → `x.y` →
/// `x`) to `g_rcutils_logging_default_logger_level`; `nros_log` has no
/// ancestry — [`get_logger`] is exact string equality over [`MAX_LOGGERS`]
/// slots — so that walk degenerates here to exactly one step, and one step is
/// what this is.
///
/// Deliberately not built further, and phase-467 recorded that as PERMANENT on
/// ledger row `c:log_severity_t` rather than as work. A dotted name is accepted
/// and is just a name with a dot in it; what is missing is the PARENT, and the
/// two things that would give it one are both absent for the same reason.
/// Upstream keys a level by NAME (`rcutils_logging_set_logger_level(const char
/// *, int)` writes an allocator-backed severity map), so an ancestor there can
/// carry a level without being a logger; here the level is an `AtomicU8` INSIDE
/// the [`Logger`], so an ancestor would have to occupy one of the 32 slots to
/// hold one. And the producer of dotted names upstream, `Logger::get_child`,
/// has no allocator here to concatenate `parent.child` with. A resolver for a
/// hierarchy nothing can construct is not a missing feature.
///
/// RFC-0102 reversed that and phase-479 built it: the premises went stale
/// ([`get_or_create_logger`] copies runtime names into a static arena,
/// `parent.child` fits a 48-byte stack buffer, and the ancestors a `get_child`
/// produces are loggers already). The hierarchy is now a parent pointer on
/// [`Logger`], linked at creation, and this byte is the LAST step of the walk
/// rather than the only one.
static DEFAULT_LEVEL: AtomicU8 = AtomicU8::new(Severity::Info as u8);

/// A named logger with a runtime severity threshold.
///
/// A logger constructed with [`Logger::new`] has NO level of its own and
/// follows [`Logger::default_level`] (itself [`Severity::Info`] until someone
/// calls [`Logger::set_default_level`]). [`Logger::with_level`] and
/// [`Logger::set_level`] give it one; [`Logger::unset_level`] takes it away
/// again. Use [`register_logger`] to publish a `'static Logger` so multiple
/// call sites with the same name share the same threshold.
pub struct Logger {
    name: &'static str,
    /// Either a [`Severity`] discriminant or the private `LEVEL_UNSET`.
    level: AtomicU8,
    /// RFC-0102 D2 — the nearest EXISTING ancestor by dotted name (`a.b` for
    /// `a.b.c`, else `a`), or null. A logger with no level of its own resolves
    /// through this chain before falling to [`Logger::default_level`], which is
    /// rcutils's effective-level rule. Set when a logger is published to the
    /// intern table ([`register_logger`], [`get_or_create_logger`]), never on
    /// the log path. Only ever points at a logger whose name is a STRICT
    /// dotted prefix of this one's, so the chain is acyclic and shorter than
    /// [`MAX_LOGGERS`].
    parent: AtomicPtr<Logger>,
    /// RFC-0102 D4 — set when a NODE's accessor resolves this logger
    /// ([`resolve_node_logger`]). The `/rosout` sink publishes by it: upstream
    /// rcl publishes only through a publisher correlated with a node's logger.
    node: AtomicBool,
}

impl Logger {
    /// `const`-construct with NO level of its own — the logger follows
    /// [`Logger::default_level`], which is [`Severity::Info`] until something
    /// moves it.
    ///
    /// phase-467: this stored `Severity::Info` outright until the process
    /// default existed, and "the author wrote `new`" then looked exactly like
    /// "the author wrote `with_level(.., Info)`". They are different
    /// statements — one is a choice and the other is the absence of one — and
    /// telling them apart is what lets [`Logger::set_default_level`] move the
    /// first without overriding the second.
    #[must_use]
    pub const fn new(name: &'static str) -> Self {
        Self {
            name,
            level: AtomicU8::new(LEVEL_UNSET),
            parent: AtomicPtr::new(core::ptr::null_mut()),
            node: AtomicBool::new(false),
        }
    }

    /// `const`-construct with an explicit threshold.
    #[must_use]
    pub const fn with_level(name: &'static str, level: Severity) -> Self {
        Self {
            name,
            level: AtomicU8::new(level as u8),
            parent: AtomicPtr::new(core::ptr::null_mut()),
            node: AtomicBool::new(false),
        }
    }

    /// Logger name (used as `Record::logger_name`).
    #[must_use]
    pub const fn name(&self) -> &'static str {
        self.name
    }

    /// The threshold this logger actually filters on — its own level if it has
    /// one, otherwise its nearest ancestor's (RFC-0102 D2), otherwise
    /// [`Logger::default_level`].
    ///
    /// This is `rcutils_logging_get_logger_effective_level`'s answer, not
    /// `rcutils_logging_get_logger_level`'s: it is the number that decides
    /// whether a record is emitted, which is the question a call site asks.
    /// There is deliberately no accessor for "is a level set on this logger",
    /// because nothing in the tree has a use for the difference and an
    /// accessor per language is three more names for one fact.
    #[must_use]
    pub fn level(&self) -> Severity {
        severity_from_u8(self.effective_level_byte()).unwrap_or(Severity::Info)
    }

    /// The nearest ancestor this logger resolves an unset level through, if
    /// any (RFC-0102 D2). `None` for a root, and for every logger not yet
    /// published to the intern table.
    #[must_use]
    pub fn parent(&self) -> Option<&'static Logger> {
        let p = self.parent.load(Ordering::Acquire);
        // SAFETY: `parent` is only ever stored from a `&'static Logger` (see
        // `link_published`), and loggers are never freed.
        if p.is_null() {
            None
        } else {
            Some(unsafe { &*p })
        }
    }

    /// Whether a node's accessor resolved this logger (RFC-0102 D4).
    #[must_use]
    pub fn is_node_logger(&self) -> bool {
        self.node.load(Ordering::Relaxed)
    }

    /// Own level byte if set; else the first set level up the parent chain;
    /// else the process default. The walk is bounded by [`MAX_LOGGERS`] even
    /// though the chain is acyclic by construction, so a defect elsewhere can
    /// cost a wrong answer but never a hang on the log path.
    #[inline]
    fn effective_level_byte(&self) -> u8 {
        let own = self.level.load(Ordering::Relaxed);
        if own != LEVEL_UNSET {
            return own;
        }
        let mut p = self.parent.load(Ordering::Acquire);
        let mut hops = 0;
        while !p.is_null() && hops < MAX_LOGGERS {
            // SAFETY: as in `parent()`.
            let ancestor: &Logger = unsafe { &*p };
            let level = ancestor.level.load(Ordering::Relaxed);
            if level != LEVEL_UNSET {
                return level;
            }
            p = ancestor.parent.load(Ordering::Acquire);
            hops += 1;
        }
        DEFAULT_LEVEL.load(Ordering::Relaxed)
    }

    /// Give this logger a level of its own. It stops following
    /// [`Logger::default_level`] until [`Logger::unset_level`].
    pub fn set_level(&self, level: Severity) {
        self.level.store(level as u8, Ordering::Relaxed);
    }

    /// Take this logger's own level away, so it follows
    /// [`Logger::default_level`] again.
    ///
    /// `rcutils_logging_set_logger_level(name, RCUTILS_LOG_SEVERITY_UNSET)`'s
    /// effect, and what `rclc_logger_set_level(logger, RCLC_LOG_SEVERITY_UNSET)`
    /// reaches. phase-467.
    ///
    /// rclrs spells this `logger.set_level(LogSeverity::Unset)`. Here it is a verb,
    /// because [`Severity`] has no `Unset`: its values are the sink ABI.
    pub fn unset_level(&self) {
        self.level.store(LEVEL_UNSET, Ordering::Relaxed);
    }

    /// The threshold every logger with no level of its own filters on.
    ///
    /// [`Severity::Info`] until [`Logger::set_default_level`] moves it, which
    /// is `RCUTILS_DEFAULT_LOGGER_DEFAULT_LEVEL`'s value and upstream's
    /// starting point too.
    #[must_use]
    pub fn default_level() -> Severity {
        severity_from_u8(DEFAULT_LEVEL.load(Ordering::Relaxed)).unwrap_or(Severity::Info)
    }

    /// Move the threshold that every logger WITHOUT a level of its own filters
    /// on — `rclrs::Logger::set_default_level`.
    ///
    /// Takes effect immediately and retroactively: a `static LOGGER:
    /// Logger = Logger::new("x")` that was compiled into the image long before
    /// this call follows it, because "no level of its own" is a stored state
    /// and not a value copied in at construction. That is the whole reason
    /// this is a capability rather than a re-export of
    /// `DEFAULT_LOGGER.set_level`, which moves the threshold of the ONE logger
    /// named `"nros"` (ledger row `rust:Logger::set_default_level`).
    ///
    /// It does NOT override a logger that states its own level: after
    /// `with_level(.., Warn)` or `set_level(Warn)`, `Warn` wins.
    ///
    /// Takes [`Severity`], not rclrs's `LogSeverity`: it has a `Trace` level rclrs
    /// lacks and no `Unset`. Where rclrs passes `Unset` to restore the default
    /// default, pass `Severity::Info`, which is what `Unset` restores upstream.
    pub fn set_default_level(level: Severity) {
        DEFAULT_LEVEL.store(level as u8, Ordering::Relaxed);
    }

    /// Whether a record at `severity` would be emitted by this
    /// logger AT RUNTIME.
    ///
    /// THE HOT PATH of every log macro, on every target, so the shape here is
    /// measured rather than chosen (phase-467; the figures below are for the
    /// two-level own-or-default form that preceded RFC-0102's ancestor walk). Resolving the unset level
    /// SELECTS the byte to compare against and compares once, instead of
    /// comparing twice and short-circuiting: measured at an inlined call site
    /// with `rustc -O`, the select form is 13 instructions on
    /// `thumbv7m-none-eabi` and **branch-free** (LLVM predicates the whole
    /// fallback into one `ittt eq` block) against 16 with a branch for the
    /// short-circuit form, and 10 against 12 on `x86_64`. The cost over the
    /// single-load original (8 thumb / 4 x86_64 instructions) is one compare
    /// against an immediate, plus a byte load of one process-global that a
    /// logger with its own level never performs.
    ///
    /// `LEVEL_UNSET` being ABOVE `Fatal` rather than below `Trace` is what
    /// keeps this correct if the select is ever reordered: `severity >=
    /// u8::MAX` is false for every `Severity`, so an unresolved sentinel can
    /// only ever suppress a record, never emit one at the most verbose level —
    /// which is the failure `0` had.
    ///
    /// RFC-0102 D2 changed the unset arm: it now walks the parent chain before
    /// the process default. A logger WITH its own level still costs one load
    /// and one compare; an unset root pays one extra pointer load; each
    /// ancestor without a level of its own costs one more. No string work, on
    /// any arm.
    #[must_use]
    ///
    /// The image's compile-time CEILING (`NROS_LOG_MAX_LEVEL`, issue 1037) is
    /// applied here too, first: a runtime level below it is unreachable. That
    /// is what brings the C and C++ surfaces under the same bound — their
    /// macros format on the C side and reach the facade through this check
    /// (`rclc_log_emit_at`, `nros_log_is_enabled`), so the ceiling the Rust
    /// macros fold away at compile time filters theirs at the FFI boundary.
    pub fn is_enabled(&self, severity: Severity) -> bool {
        severity_enabled_at_compile_time(severity)
            && (severity as u8) >= self.effective_level_byte()
    }

    /// Whether a record at `severity` would be emitted, given both this
    /// logger's runtime threshold AND a per-call-site throttle window.
    ///
    /// phase-417 W4.d. The order is load-bearing and is the same order
    /// `rclcpp` uses: the SEVERITY test runs first, so a record the level
    /// already filters does not consume the throttle window. Reversing them
    /// makes a raised-then-lowered log level lose its first record for a whole
    /// interval, which is the shape of bug nobody reports because it looks
    /// like the throttle working.
    ///
    /// Consumes the window when it returns `true`, so call it exactly once per
    /// candidate record — the `nros_*_throttle!` macros do.
    ///
    /// `now_ns` is the caller's monotonic clock; the macros pass
    /// [`__timestamp_ns`]. Taking it as an argument rather than reading the
    /// clock here is what lets the whole rule be tested without a platform
    /// port, and is the same shape `rclcpp::RCLCPP_*_THROTTLE` uses when it
    /// asks for a `Clock`.
    #[must_use]
    pub fn is_enabled_throttled(
        &self,
        severity: Severity,
        state: &ThrottleState,
        now_ns: u64,
        interval_ns: u64,
    ) -> bool {
        self.is_enabled(severity) && state.should_log(now_ns, interval_ns)
    }

    /// Hand `record` to every registered sink, after the runtime
    /// threshold check.
    ///
    /// Macros call this; user code should not.
    pub fn dispatch(&self, record: &Record<'_>) {
        if !self.is_enabled(record.severity) {
            return;
        }
        dispatch_to_sinks(record);
    }
}

// -----------------------------------------------------------------------------
// Static intern table for `get_logger("name")`. Bounded; no alloc.
// -----------------------------------------------------------------------------

/// Maximum number of named loggers that can be registered via
/// [`register_logger`]. Beyond this, [`get_logger`] returns
/// [`DEFAULT_LOGGER`].
pub const MAX_LOGGERS: usize = 32;

/// Catch-all logger returned when the requested name is not
/// registered (or the intern table is full).
pub static DEFAULT_LOGGER: Logger = Logger::new("nros");

mod intern {
    use super::{AtomicPtr, Logger, MAX_LOGGERS, Ordering};

    pub(super) struct InternTable {
        slots: [AtomicPtr<Logger>; MAX_LOGGERS],
    }

    impl InternTable {
        /// Every published logger, in slot order.
        pub(super) fn iter(&self) -> impl Iterator<Item = &'static Logger> + '_ {
            self.slots.iter().map_while(|slot| {
                let ptr = slot.load(Ordering::Acquire);
                // SAFETY: same publication invariant as `lookup`.
                if ptr.is_null() {
                    None
                } else {
                    Some(unsafe { &*ptr })
                }
            })
        }

        pub(super) const fn new() -> Self {
            // `AtomicPtr::new` is `const` on both `core::sync::atomic`
            // and `portable_atomic`, so we can initialise the array
            // by repeating the call rather than naming a `const` —
            // which clippy flags as interior-mutable.
            #[allow(clippy::declare_interior_mutable_const)]
            const NULL: AtomicPtr<Logger> = AtomicPtr::new(core::ptr::null_mut());
            Self {
                slots: [NULL; MAX_LOGGERS],
            }
        }

        pub(super) fn lookup(&self, name: &str) -> Option<&'static Logger> {
            for slot in &self.slots {
                let ptr = slot.load(Ordering::Acquire);
                if ptr.is_null() {
                    return None;
                }
                // SAFETY: pointer published via Release after the
                // owner constructed a `'static Logger`. The Acquire
                // load synchronizes.
                let logger: &'static Logger = unsafe { &*ptr };
                if logger.name() == name {
                    return Some(logger);
                }
            }
            None
        }

        pub(super) fn insert(&self, logger: &'static Logger) -> Option<&'static Logger> {
            if let Some(existing) = self.lookup(logger.name()) {
                return Some(existing);
            }
            let ptr = logger as *const _ as *mut Logger;
            for slot in &self.slots {
                if slot
                    .compare_exchange(
                        core::ptr::null_mut(),
                        ptr,
                        Ordering::AcqRel,
                        Ordering::Acquire,
                    )
                    .is_ok()
                {
                    return Some(logger);
                }
                let existing_ptr = slot.load(Ordering::Acquire);
                if !existing_ptr.is_null() {
                    // SAFETY: same publication invariant as `lookup`.
                    let existing: &'static Logger = unsafe { &*existing_ptr };
                    if existing.name() == logger.name() {
                        return Some(existing);
                    }
                }
            }
            None
        }
    }
}

static INTERN: intern::InternTable = intern::InternTable::new();

/// The dotted ancestors of `name`, nearest first: `a.b.c` yields `a.b`, `a`.
fn dotted_ancestors(name: &str) -> impl Iterator<Item = &str> {
    let mut rest = name;
    core::iter::from_fn(move || {
        let cut = rest.rfind('.')?;
        rest = &rest[..cut];
        Some(rest)
    })
}

/// RFC-0102 D2 — wire a logger that was JUST published into the hierarchy.
///
/// 1. Its parent is its nearest EXISTING dotted ancestor.
/// 2. Every published logger whose current parent is a FARTHER ancestor than
///    `new` (or which has none) and whose name has `new`'s as a dotted prefix
///    is re-pointed at `new`. So the result does not depend on creation order:
///    `a.b.c` created before `a.b` still resolves through `a.b` once it exists.
///
/// Runs on publication only — one pass over at most [`MAX_LOGGERS`] slots.
/// Creation is expected at initialisation; two loggers created at the same
/// instant from different threads can leave a link one level FARTHER than
/// ideal, never one pointing at a non-ancestor, because every store here is of
/// a logger whose name is a strict dotted prefix of the target's.
fn link_published(new: &'static Logger) {
    let ptr = new as *const Logger as *mut Logger;
    if let Some(parent) = dotted_ancestors(new.name()).find_map(|a| INTERN.lookup(a)) {
        new.parent
            .store(parent as *const Logger as *mut Logger, Ordering::Release);
    }
    let n = new.name();
    for other in INTERN.iter() {
        let on = other.name();
        let is_descendant =
            on.len() > n.len() && on.starts_with(n) && on.as_bytes()[n.len()] == b'.';
        if !is_descendant {
            continue;
        }
        let closer = match other.parent() {
            None => true,
            Some(current) => current.name().len() < n.len(),
        };
        if closer {
            other.parent.store(ptr, Ordering::Release);
        }
    }
}

/// Publish through the intern table and, when THIS call did the inserting,
/// link the logger into the hierarchy. The one spelling both publication
/// paths use.
fn publish(logger: &'static Logger) -> Option<&'static Logger> {
    let published = INTERN.insert(logger)?;
    if core::ptr::eq(published, logger) {
        link_published(published);
    }
    Some(published)
}

/// Publish `logger` under its name so subsequent `get_logger`
/// calls with that name return THIS reference.
///
/// On name collision returns the pre-existing entry (the input
/// `logger` is NOT inserted). On a full table returns
/// [`DEFAULT_LOGGER`].
pub fn register_logger(logger: &'static Logger) -> &'static Logger {
    publish(logger).unwrap_or(&DEFAULT_LOGGER)
}

/// Look up a registered logger by name. Returns [`DEFAULT_LOGGER`]
/// if none is registered (call [`register_logger`] for a `'static
/// Logger` to publish one).
///
/// Total — every call returns a usable handle the macros can
/// dispatch through.
#[must_use]
pub fn get_logger(name: &str) -> &'static Logger {
    INTERN.lookup(name).unwrap_or(&DEFAULT_LOGGER)
}

/// Look up a logger by name, CREATING one if the name is new.
///
/// phase-417 W4.d. This is the shape `rclcpp::get_logger` / `rcutils`' logger
/// lookup have, and the one a wrapper needs: [`get_logger`] answers
/// [`DEFAULT_LOGGER`] for an unregistered name, so a C or C++ caller doing
/// `set_level(get_logger("nav"), Debug)` over the lookup-only form would have
/// moved the threshold of the catch-all logger that EVERY other unregistered
/// name also resolves to. Same call, same types, different effect — precisely
/// the compile-and-differ RFC-0089 forbids.
///
/// Storage is the bounded static arena in [`pool`], sized by the
/// `NROS_LOG_DYNAMIC_LOGGERS` knob (RFC-0102 D5: env > board > 16). Returns `None` — never a logger under the
/// wrong name — when:
///
/// * `name` is empty or longer than [`MAX_LOGGER_NAME_LEN`],
/// * the name arena or the logger arena is full,
/// * the [`MAX_LOGGERS`] intern table is full.
///
/// Callers decide what an exhausted arena means for them; `nros-c` reports it
/// and falls back to [`DEFAULT_LOGGER`] loudly rather than silently.
///
/// Loggers created this way are never destroyed — the arena has no free list,
/// which is what makes the returned `&'static Logger` honest on `no_std`.
#[must_use]
pub fn get_or_create_logger(name: &str) -> Option<&'static Logger> {
    if let Some(existing) = INTERN.lookup(name) {
        return Some(existing);
    }
    let interned = pool::intern_name(name)?;
    let placed = pool::place(Logger::new(interned))?;
    // Not `register_logger`: that answers `DEFAULT_LOGGER` on a full table,
    // which would hand back a logger under the wrong name — the exact aliasing
    // this function exists to avoid. On a lost race `insert` returns the
    // winner and our slot is spent, which `pool::dynamic_loggers_in_use`
    // reports.
    publish(placed)
}

/// Why [`Logger::create_child`] made no child — RFC-0102 D1.
///
/// rclrs's `create_child` is fallible too (`Result<Logger, RclrsError>`), so a
/// Rust caller chooses its own fallback (`parent.create_child("x").unwrap_or(parent)`).
/// The C and C++ surfaces, whose upstream shapes cannot report failure, fall
/// back to the parent automatically (RFC-0102 D3).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChildError {
    /// The child's full name is longer than [`MAX_LOGGER_NAME_LEN`] bytes.
    NameTooLong {
        /// Bytes the full name would have.
        len: usize,
        /// [`MAX_LOGGER_NAME_LEN`].
        max: usize,
    },
    /// The full name would be empty: an empty `child_name` under the
    /// catch-all, whose children are top-level names.
    EmptyName,
    /// The runtime-logger arena, its name bytes, or the intern table is full.
    ArenaFull,
}

impl core::fmt::Display for ChildError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::NameTooLong { len, max } => {
                write!(
                    f,
                    "child logger name is {len} bytes, over the {max}-byte limit"
                )
            }
            Self::EmptyName => f.write_str("child logger name is empty"),
            Self::ArenaFull => f.write_str(
                "no runtime logger slot left (arena, name bytes, or intern table is full)",
            ),
        }
    }
}

impl Logger {
    /// A logger that is a descendant of this one — rclrs's `create_child`,
    /// rclcpp's `get_child` (RFC-0102 D1).
    ///
    /// Named `"{self}.{child_name}"`, except that the catch-all
    /// ([`DEFAULT_LOGGER`]) has top-level children: its child is named
    /// `child_name`, as rclrs does for its empty-named default logger.
    ///
    /// A child with no level of its own filters at its nearest ancestor's
    /// (rcutils's rule, RFC-0102 D2), so `parent.set_level(Debug)` reaches it.
    /// Idempotent: the same name returns the same logger, so a call inside a
    /// log statement costs one intern-table lookup.
    ///
    /// Takes `&'static self` because the child links to its parent by
    /// reference. `self` is published to the intern table first if it was
    /// not, so a `static` parent nobody registered still passes its level on.
    ///
    /// # Errors
    /// [`ChildError`] when the full name is over [`MAX_LOGGER_NAME_LEN`] or
    /// empty, or no slot is left. The name is built in a stack buffer; nothing
    /// is allocated.
    pub fn create_child(
        &'static self,
        child_name: impl core::borrow::Borrow<str>,
    ) -> Result<&'static Logger, ChildError> {
        let child = child_name.borrow();
        let top = core::ptr::eq(self, &DEFAULT_LOGGER);
        let len = if top {
            child.len()
        } else {
            self.name.len() + 1 + child.len()
        };
        if len == 0 {
            return Err(ChildError::EmptyName);
        }
        if len > MAX_LOGGER_NAME_LEN {
            return Err(ChildError::NameTooLong {
                len,
                max: MAX_LOGGER_NAME_LEN,
            });
        }
        if !top && INTERN.lookup(self.name).is_none() {
            // A full table leaves the child unlinked: it then filters at the
            // process default, which is what an unregistered parent meant
            // before RFC-0102 too.
            let _ = publish(self);
        }
        let mut buf = [0u8; MAX_LOGGER_NAME_LEN];
        let mut at = 0;
        if !top {
            buf[..self.name.len()].copy_from_slice(self.name.as_bytes());
            at = self.name.len();
            buf[at] = b'.';
            at += 1;
        }
        buf[at..len].copy_from_slice(child.as_bytes());
        // SAFETY: two `&str`s joined by an ASCII `.` are valid UTF-8, and the
        // cut is at their end.
        let name = unsafe { core::str::from_utf8_unchecked(&buf[..len]) };
        get_or_create_logger(name).ok_or(ChildError::ArenaFull)
    }
}

/// The published logger called `name`, if any — the lookup the `/rosout`
/// sink filters by (RFC-0102 D4), which sees a record's name, not its logger.
#[cfg(feature = "rosout")]
pub(crate) fn find_published(name: &str) -> Option<&'static Logger> {
    INTERN.lookup(name)
}

/// [`resolve_logger`] for a NODE's own logger, marking it as one — the one
/// spelling the node-shaped accessors use (`Node::logger`,
/// `rclc_node_get_logger`, the C++ node's), so the `/rosout` sink can publish
/// what upstream publishes (RFC-0102 D4). The catch-all is never marked: a node
/// whose name fell back to it would otherwise publish every unnamed logger.
#[must_use]
pub fn resolve_node_logger(name: &str) -> &'static Logger {
    let logger = resolve_logger(name);
    if !core::ptr::eq(logger, &DEFAULT_LOGGER) {
        logger.node.store(true, Ordering::Relaxed);
    }
    logger
}

/// Resolve `name` to a logger — the TOTAL form of [`get_or_create_logger`],
/// and the one spelling every node-shaped accessor uses.
///
/// phase-427 W5. Four accessors answer "which logger is this node's?" —
/// `nros_node::executor::Node::logger`, `rclc_node_get_logger`,
/// `nros_cpp_node_get_logger` and `rclc_log_get_logger`. All four had the same
/// job and, before this, three of them called [`get_logger`], which is a
/// LOOKUP: a node whose name nobody `register_logger`ed resolved to
/// [`DEFAULT_LOGGER`]. Two nodes in one image therefore emitted every record
/// under the same name, `"nros"` — the sentinel W5 replaced, arrived at by a
/// different route.
///
/// This creates instead, and falls back to [`DEFAULT_LOGGER`] only when it
/// cannot — an empty or over-long name, or an exhausted arena — and says so at
/// WARN when it does, because the catch-all's threshold is SHARED and a silent
/// alias makes `set_level` on the result move every other unnamed logger's bar
/// with it. The report is ONE per process: a C++ `NROS_LOG_*` call site asks
/// `node.get_logger()` on every line, so a per-call report would be a flood.
///
/// Lookup comes first, so an image that pre-registered a `'static Logger` under
/// the node's name still gets that exact reference and its threshold.
#[must_use]
pub fn resolve_logger(name: &str) -> &'static Logger {
    if let Some(logger) = get_or_create_logger(name) {
        return logger;
    }
    // ONE report per process. This is reached from `Node::get_logger()`, which
    // a C++ call site invokes inside every `NROS_LOG_*` macro — so a node whose
    // name is longer than `MAX_LOGGER_NAME_LEN` (node names may be 64 bytes,
    // logger names 48) would otherwise emit a warning per log line, and the
    // diagnostic would be the flood. The first failure is the informative one:
    // the arena only ever gets fuller.
    static REPORTED: AtomicU8 = AtomicU8::new(0);
    if REPORTED.swap(1, Ordering::Relaxed) != 0 {
        return &DEFAULT_LOGGER;
    }
    let (used, total) = dynamic_logger_name_arena();
    crate::log_warn!(
        &DEFAULT_LOGGER,
        "resolve_logger(\"{name}\"): no logger created — {} of {} slots and {used} of {total} \
         name bytes are spent, or the name is over {} bytes. Returning the catch-all logger, \
         whose threshold is SHARED: raise NROS_LOG_DYNAMIC_LOGGERS rather than calling \
         set_level on this handle. Reported once per process.",
        dynamic_loggers_in_use(),
        dynamic_logger_capacity(),
        MAX_LOGGER_NAME_LEN,
    );
    &DEFAULT_LOGGER
}

// -----------------------------------------------------------------------------
// Sink list. Set once at `init`; read every dispatch.
// -----------------------------------------------------------------------------

static SINKS_PTR: AtomicPtr<&'static [&'static dyn LogSink]> =
    AtomicPtr::new(core::ptr::null_mut());

// issue 0710 — `init_default()` was here. It named a default this crate can no
// longer name: the platform sink moved to `nros_platform_cffi::log`, where the
// ABI it speaks is a dependency rather than a feature. Call
// `nros_platform_cffi::log::init_default()`, or `init()` with your own sinks.

/// Install the global sink list.
///
/// MUST be called at app startup BEFORE any record-emitting macro
/// runs (otherwise the dispatch is a no-op — records are silently
/// dropped). Calling `init` more than once swaps the list
/// atomically; the previous pointer is leaked (intentional: the
/// read path is lock-free so we can't safely free).
///
/// The sinks themselves must outlive the program (`'static`).
pub fn init(sinks: &'static [&'static dyn LogSink]) {
    // Indirect through a small `'static` cell so the read path
    // dereferences a fat-pointer-sized slot rather than reading
    // a wide pointer atomically.
    #[cfg(feature = "alloc")]
    {
        let boxed: alloc::boxed::Box<&'static [&'static dyn LogSink]> =
            alloc::boxed::Box::new(sinks);
        let ptr = alloc::boxed::Box::into_raw(boxed);
        SINKS_PTR.store(ptr, Ordering::Release);
    }
    #[cfg(not(feature = "alloc"))]
    {
        static CELL: SinkSlot = SinkSlot::new();
        CELL.store(sinks);
        SINKS_PTR.store(CELL.as_ptr(), Ordering::Release);
    }
    // AFTER publishing, so a record raised during the drain reaches `sinks`
    // directly rather than joining a ring nobody will drain again. Overflow is
    // reported by the same path (`report_lost`).
    deliver_held();
}

#[cfg(not(feature = "alloc"))]
struct SinkSlot {
    inner: core::cell::UnsafeCell<Option<&'static [&'static dyn LogSink]>>,
}

#[cfg(not(feature = "alloc"))]
// SAFETY: only written from `init`, which the user contracts to call
// once at startup before any concurrent reader exists.
unsafe impl Sync for SinkSlot {}

#[cfg(not(feature = "alloc"))]
impl SinkSlot {
    const fn new() -> Self {
        Self {
            inner: core::cell::UnsafeCell::new(None),
        }
    }
    fn store(&self, sinks: &'static [&'static dyn LogSink]) {
        // SAFETY: see Sync note above.
        unsafe {
            *self.inner.get() = Some(sinks);
        }
    }
    fn as_ptr(&self) -> *mut &'static [&'static dyn LogSink] {
        self.inner.get().cast()
    }
}

// -----------------------------------------------------------------------------
// Appendable sink registry (phase-417 W4.d).
//
// `init` REPLACES the list, which is right for the board — it names, once, the
// whole delivery it chose. It is the wrong verb for a consumer: C and C++ had
// no way to install a sink at all (`rclc_log_init(void)` takes no arguments
// and hardcodes the platform default), and a library that wants to tee records
// to /rosout cannot call `init` without discarding whatever the board
// installed.
//
// So `add_sink` APPENDS, and dispatch walks both lists. The two are kept
// separate rather than merged into one growable list because they have
// different owners and different lifetimes-of-decision: `init` may be called
// again and swap the board's choice wholesale, and doing so must not silently
// unregister a consumer's sink.
// -----------------------------------------------------------------------------

/// How many sinks [`add_sink`] can append on top of the [`init`] list.
///
/// Small on purpose: the appendable list is for consumers that TEE (a /rosout
/// bridge, a test collector), not for composing a delivery — a board composing
/// its own delivery passes the whole slice to [`init`].
pub const MAX_ADDED_SINKS: usize = 4;

struct AddedSinks {
    slots: core::cell::UnsafeCell<[Option<&'static dyn LogSink>; MAX_ADDED_SINKS]>,
    /// Number of slots fully written. `Release` on publish, `Acquire` on read.
    len: AtomicUsize,
}

// SAFETY: writes happen only under `ADD_LOCK` and only to the slot at index
// `len`, which no reader looks at until the `Release` store to `len` that
// follows the write is observed by an `Acquire` load. A slot is written once
// and never rewritten.
unsafe impl Sync for AddedSinks {}

static ADDED: AddedSinks = AddedSinks {
    slots: core::cell::UnsafeCell::new([None; MAX_ADDED_SINKS]),
    len: AtomicUsize::new(0),
};

/// Serialises `add_sink` against itself.
///
/// A spin lock, held for two stores with no call inside it, is admissible here
/// where the in-order-publish alternative is not: that one has a thread
/// waiting on ANOTHER thread's progress, which on a single-core RTOS with
/// priorities is a deadlock rather than a delay. Do not call `add_sink` from
/// an ISR.
static ADD_LOCK: AtomicBool = AtomicBool::new(false);

/// Append `sink` to the sinks every record is delivered to.
///
/// Returns `false` when [`MAX_ADDED_SINKS`] are already registered — the sink
/// is NOT installed, and the caller is expected to say so rather than carry on
/// as if it were.
///
/// Records raised before any sink existed are replayed into `sink` here, the
/// same way [`init`] replays them, so the first sink to arrive still sees the
/// boot story regardless of which of the two calls installed it.
pub fn add_sink(sink: &'static dyn LogSink) -> bool {
    while ADD_LOCK
        .compare_exchange(false, true, Ordering::Acquire, Ordering::Acquire)
        .is_err()
    {
        core::hint::spin_loop();
    }
    let idx = ADDED.len.load(Ordering::Relaxed);
    let installed = if idx < MAX_ADDED_SINKS {
        // SAFETY: `ADD_LOCK` makes this the only writer, `idx` is beyond what
        // any reader may look at until the `Release` store below publishes it,
        // and the slot is never written again.
        unsafe {
            ADDED
                .slots
                .get()
                .cast::<Option<&'static dyn LogSink>>()
                .add(idx)
                .write(Some(sink));
        }
        ADDED.len.store(idx + 1, Ordering::Release);
        true
    } else {
        false
    };
    ADD_LOCK.store(false, Ordering::Release);

    if installed {
        // The first delivery target in the program replays what was held
        // before it. AFTER publishing, for the reason `init` states.
        deliver_held();
    }
    installed
}

/// How many sinks [`add_sink`] has installed.
#[must_use]
pub fn added_sink_count() -> usize {
    ADDED.len.load(Ordering::Acquire)
}

/// Hand `record` to every sink installed by [`add_sink`]. Returns how many
/// saw it.
fn deliver_to_added(record: &Record<'_>) -> usize {
    let len = ADDED.len.load(Ordering::Acquire);
    let base = ADDED.slots.get().cast::<Option<&'static dyn LogSink>>();
    for idx in 0..len {
        // SAFETY: `idx < len`, and the `Acquire` load above pairs with the
        // `Release` store in `add_sink` that published slot `idx`. Slots below
        // `len` are immutable for the rest of the program.
        if let Some(sink) = unsafe { base.add(idx).read() } {
            sink.log(record);
        }
    }
    len
}

/// Whether [`__timestamp_ns`] can answer with a real clock.
///
/// `false` means every `Record::timestamp_ns` is `0` and anything DERIVED from
/// the timestamp — the throttle windows above all of them — has no time base:
/// a throttled site then emits its FIRST record and nothing after it (a clock
/// that never moves is a window that never elapses — issue 1152). Exposed so a
/// wrapper can say so out loud instead of degrading quietly: without this, a
/// throttled C call site looks identical whether it is rate-limiting or
/// permanently shut.
#[must_use]
pub const fn timestamp_available() -> bool {
    cfg!(nros_log_clock)
}

/// Current monotonic time for `Record::timestamp_ns` (issue #503).
///
/// With the platform clock (`cfg(nros_log_clock)`: the lane's platform
/// declares `[capabilities] clock = true`, or the `platform-clock` feature
/// pulled a port in — issue 1037, see `build.rs`) this reads
/// `nros_platform_clock_ns` — the universal per-platform export the
/// executor's timer accounting already links. Without it it returns `0`
/// ("unavailable") and imposes no link-time requirement.
///
/// Public because the emission macros expand it in user crates; not
/// part of the supported API surface.
#[doc(hidden)]
#[must_use]
pub fn __timestamp_ns() -> u64 {
    #[cfg(nros_log_clock)]
    {
        unsafe extern "C" {
            fn nros_platform_clock_ns() -> u64;
        }
        // SAFETY: bare query of the platform's monotonic us counter;
        // the symbol comes from whichever `nros-platform-<rtos>` port
        // linked the binary (the contract `PlatformSink ->
        // nros_platform_log_write` already relies on).
        unsafe { nros_platform_clock_ns() }
    }
    #[cfg(not(nros_log_clock))]
    {
        0
    }
}

fn dispatch_to_sinks(record: &Record<'_>) {
    if !has_destination() {
        // Nothing installed yet — HOLD the record; `init` replays it into
        // whatever sinks the board chooses. See `early` for why this crate
        // does not reach for the platform sink itself: doing that (issue 0710)
        // put `nros_platform_log_write` on a path every binary executes, which
        // turned a pluggable delivery into a link-time requirement that no
        // Cargo feature can undo under workspace feature unification.
        //
        // It is also a better answer to issue 0708 than installing a default
        // was: the early records land in the sink the board picked, not in the
        // one dispatch guessed before the board had spoken.
        early::hold(record);
        return;
    }
    if !dispatch_guard_try_acquire() {
        // Another delivery is in progress — this thread's own sink logging
        // (recursion), or another thread. Hold it, then take the hand-off
        // (`early` module docs, issue 2000).
        early::hold(record);
        if dispatch_guard_try_acquire() {
            drain_and_release();
        }
        return;
    }
    deliver(record);
    drain_and_release();
}

/// Whether a delivery would reach any sink.
///
/// phase-417 W4.d — the appendable list counts as much as `init`'s: a record
/// with at least one destination is NOT held.
fn has_destination() -> bool {
    !SINKS_PTR.load(Ordering::Acquire).is_null() || ADDED.len.load(Ordering::Acquire) > 0
}

/// Hand `record` to every sink, appended ones first. Called with the dispatch
/// guard held.
fn deliver(record: &Record<'_>) {
    deliver_to_added(record);
    let ptr = SINKS_PTR.load(Ordering::Acquire);
    if !ptr.is_null() {
        // SAFETY: `init` published a valid `'static` slice reference.
        let sinks: &'static [&'static dyn LogSink] = unsafe { *ptr };
        for sink in sinks {
            sink.log(record);
        }
    }
}

/// Deliver whatever is held, if a sink exists and no delivery is in progress.
/// One in progress delivers it itself before it lets go.
fn deliver_held() {
    // A loss is worth a delivery on its own: `NROS_LOG_EARLY_RECORDS = 0`
    // holds nothing, so the count may be all there is to report.
    if has_destination() && (early::pending() || lost_unreported()) && dispatch_guard_try_acquire()
    {
        drain_and_release();
    }
}

/// With the guard held: deliver the held records, release, and take the guard
/// back while a record is still waiting — at most [`early::DRAIN_PASSES`]
/// times, which is what bounds a sink that logs (`early` module docs).
fn drain_and_release() {
    for pass in 0..early::DRAIN_PASSES {
        early::drain_with(&mut deliver);
        report_lost();
        dispatch_guard_release();
        // `SeqCst` on both: a raiser that saw the guard held published its
        // record before retrying the guard, so either it gets the guard or
        // this load sees the record.
        if pass + 1 == early::DRAIN_PASSES || !early::pending() || !dispatch_guard_try_acquire() {
            break;
        }
    }
}

/// Records lost to a full hold queue, already reported. Written only with the
/// dispatch guard held.
static LOST_REPORTED: AtomicUsize = AtomicUsize::new(0);

fn lost_unreported() -> bool {
    early::overflowed() != LOST_REPORTED.load(Ordering::Relaxed)
}

/// With the guard held: say how many records were lost since the last report.
///
/// `nros-log` cannot know what those records said, but it does know how many
/// there were, and a silent hole is worse than a count. Delivered directly
/// rather than through dispatch, because a report that could itself be held —
/// and so lost — would undercount the loss it exists to state.
fn report_lost() {
    let lost = early::overflowed();
    let reported = LOST_REPORTED.swap(lost, Ordering::Relaxed);
    if lost == reported || !severity_enabled_at_compile_time(Severity::Warn) {
        return;
    }
    let mut message = buffer::FormatBuffer::new();
    let _ = core::fmt::Write::write_fmt(
        &mut message,
        format_args!(
            "{} log record(s) lost: more were waiting for a sink than \
             NROS_LOG_EARLY_RECORDS = {} holds (see `nros_log::early`)",
            lost - reported,
            early::early_depth()
        ),
    );
    deliver(&Record {
        severity: Severity::Warn,
        logger_name: "nros_log",
        message: message.as_str(),
        file: file!(),
        line: line!(),
        timestamp_ns: __timestamp_ns(),
    });
}

/// Flush every registered sink, appended ones included.
pub fn flush() {
    // A record held for a busy dispatcher (issue 2000) is not flushed until it
    // has been delivered.
    deliver_held();
    let len = ADDED.len.load(Ordering::Acquire);
    let base = ADDED.slots.get().cast::<Option<&'static dyn LogSink>>();
    for idx in 0..len {
        // SAFETY: same publication invariant as `deliver_to_added`.
        if let Some(sink) = unsafe { base.add(idx).read() } {
            sink.flush();
        }
    }
    let ptr = SINKS_PTR.load(Ordering::Acquire);
    if ptr.is_null() {
        return;
    }
    // SAFETY: same invariant as `dispatch_to_sinks`.
    let sinks: &'static [&'static dyn LogSink] = unsafe { *ptr };
    for sink in sinks {
        sink.flush();
    }
}

// -----------------------------------------------------------------------------
// Dispatch guard — one delivery at a time, process-wide.
//
// Held across every `sink.log()`. It breaks a sink that logs from recursing,
// and it keeps two threads out of the sinks at once, so a sink never sees
// interleaved calls. A record that finds it held is NOT dropped (issue 2000):
// it is held in `early` and delivered by the holder. Per-thread recursion
// detection would need a thread identity, which no `no_std` target here gives
// for free; deferral needs none.
// -----------------------------------------------------------------------------

static DISPATCH_GUARD: AtomicBool = AtomicBool::new(false);

fn dispatch_guard_try_acquire() -> bool {
    DISPATCH_GUARD
        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
        .is_ok()
}

fn dispatch_guard_release() {
    DISPATCH_GUARD.store(false, Ordering::SeqCst);
}

// -----------------------------------------------------------------------------
// Tests (host-only).
// -----------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn severity_round_trips_through_u8() {
        for s in [
            Severity::Trace,
            Severity::Debug,
            Severity::Info,
            Severity::Warn,
            Severity::Error,
            Severity::Fatal,
        ] {
            assert_eq!(severity_from_u8(s.as_u8()), Some(s));
        }
        assert_eq!(severity_from_u8(99), None);
    }

    #[test]
    fn logger_runtime_threshold_filters_below() {
        let logger = Logger::with_level("test_thresh", Severity::Warn);
        assert!(!logger.is_enabled(Severity::Info));
        assert!(logger.is_enabled(Severity::Warn));
        assert!(logger.is_enabled(Severity::Error));
        logger.set_level(Severity::Debug);
        assert!(logger.is_enabled(Severity::Info));
    }

    /// phase-467, ledger rows `c:log_severity_t` + `rust:Logger::set_default_level`.
    ///
    /// ONE test rather than five, deliberately: the process default is
    /// process-wide state and `cargo test` runs this crate's tests as threads
    /// in one process, so a second test that moved it would race this one and
    /// the flake would read as a logic bug. It restores the default before it
    /// returns for the same reason.
    #[test]
    fn a_logger_with_no_level_of_its_own_follows_the_process_default() {
        assert_eq!(Logger::default_level(), Severity::Info, "starting default");

        let unset = Logger::new("test_unset_follows_default");
        let stated = Logger::with_level("test_stated_keeps_its_own", Severity::Warn);

        // Unset logger, untouched default: Info, exactly as before phase-467.
        assert_eq!(unset.level(), Severity::Info);
        assert!(!unset.is_enabled(Severity::Debug));
        assert!(unset.is_enabled(Severity::Info));

        // Moving the default moves it RETROACTIVELY — the logger already
        // existed. This is what a re-export of `DEFAULT_LOGGER.set_level`
        // could not do.
        Logger::set_default_level(Severity::Debug);
        assert_eq!(Logger::default_level(), Severity::Debug);
        assert_eq!(unset.level(), Severity::Debug);
        assert!(unset.is_enabled(Severity::Debug));
        assert!(!unset.is_enabled(Severity::Trace));

        // ...and does NOT override a logger that states its own level. If it
        // did, `with_level(.., Info)` and `new(..)` would be the same thing,
        // which is the distinction the sentinel exists to keep.
        assert_eq!(stated.level(), Severity::Warn);
        assert!(!stated.is_enabled(Severity::Info));

        // Stating a level on the unset logger detaches it.
        unset.set_level(Severity::Error);
        Logger::set_default_level(Severity::Trace);
        assert_eq!(unset.level(), Severity::Error);
        assert!(!unset.is_enabled(Severity::Warn));

        // ...and unsetting it re-attaches it, to the CURRENT default.
        unset.unset_level();
        assert_eq!(unset.level(), Severity::Trace);
        assert!(unset.is_enabled(Severity::Trace));

        // The sentinel is outside `Severity`, in the direction that can only
        // ever SUPPRESS a record: `severity_from_u8` refuses it, and it is
        // above `Fatal`, not below `Trace` where rcutils's own `UNSET` sits.
        assert_eq!(severity_from_u8(LEVEL_UNSET), None);
        assert!(LEVEL_UNSET > Severity::Fatal.as_u8());

        Logger::set_default_level(Severity::Info);
    }

    #[test]
    fn unregistered_get_logger_returns_default() {
        let l = get_logger("definitely-not-registered-99");
        assert_eq!(l.name(), DEFAULT_LOGGER.name());
    }

    #[test]
    fn registered_logger_round_trips_through_intern_table() {
        static LOGGER: Logger = Logger::new("test_intern_round_trip");
        let published = register_logger(&LOGGER);
        assert_eq!(published.name(), LOGGER.name());
        let looked_up = get_logger("test_intern_round_trip");
        assert!(core::ptr::eq(published, looked_up));
    }

    #[test]
    fn compile_time_ceiling_matches_the_resolved_knob() {
        // `config::MAX_LEVEL` is `NROS_LOG_MAX_LEVEL` as `build.rs` resolved
        // it (builtin 0 = trace; 6 = off).
        let ceiling = severity_from_u8(config::MAX_LEVEL);
        for severity in [
            Severity::Trace,
            Severity::Debug,
            Severity::Info,
            Severity::Warn,
            Severity::Error,
            Severity::Fatal,
        ] {
            let enabled = ceiling.is_some_and(|c| severity >= c);
            assert_eq!(severity_enabled_at_compile_time(severity), enabled);
            // The ceiling bounds the runtime level: below it, no logger level
            // can switch a severity back on.
            static PROBE: Logger = Logger::with_level("ceiling_probe", Severity::Trace);
            assert_eq!(PROBE.is_enabled(severity), enabled);
        }
    }
}