nros_node/executor/

arena.rs

//! Callback arena infrastructure (all pub(crate)).

use core::marker::PhantomData;

use nros_core::{
    CdrReader, DeserializeView, MessageInfo, RawMessageInfo, RosAction, RosMessage, RosService,
    ViewableMessage,
};
use nros_rmw::{ServiceTrait, Subscription, TransportError};

use super::{
    action_core::{ActionClientCore, ActionServerCore},
    handles::{ActionServer, ActiveGoal},
    spsc_ring::SpscRing,
    triple_buffer::TripleBuffer,
    types::{
        InvocationMode, NodeError, RawAcceptedCallback, RawCancelCallback,
        RawCancelResponseCallback, RawFeedbackCallback, RawGoalCallback, RawGoalResponseCallback,
        RawMessageDeserializeFn, RawRequestId, RawResponseCallback, RawResponseRequestIdCallback,
        RawResultCallback, RawServiceCallback, RawServiceRequestIdCallback,
        RawSubscriptionCallback, RawSubscriptionInfoCallback, TypedSubscriptionCallback,
    },
};
use crate::session;
// phase-430 W4 — the clock-source enum moved to the ungated `timer` module so
// the declarative metadata layer can name it; `executor` re-exports it.
pub(crate) use crate::timer::{TimerCallInfo, TimerClockSource};

// ============================================================================
// Callback metadata
// ============================================================================

// ============================================================================
// Phase 8 — callback dispatch hooks (paired stubs)
// ============================================================================
//
// `docs/design/callback_tracing.rst`. The hooks bracket the LEAF callback
// invocation rather than the `try_process` boundary, because that boundary
// gets both granularity and truth wrong: one `try_process` for an action
// server fires up to three distinct user callbacks, a ring-buffered
// subscription fires the user callback once per queued message, and
// `Ok(false)` — "ran, fired nothing", the common outcome for a timer that is
// not yet due — would be recorded as an invocation that never happened.
//
// Paired stubs (the `entry_tiers.rs` idiom) so the call sites read IDENTICALLY
// whether or not the feature is on; the `#[cfg]` lives here, once, instead of
// wrapping every hook site.

/// Emit callback_start(handle) immediately before invoking a user callback.
#[cfg(feature = "trace-callbacks")]
#[inline]
fn trace_cb_start(desc_idx: u8) {
    super::callback_trace::start(desc_idx);
}

#[cfg(not(feature = "trace-callbacks"))]
#[inline]
fn trace_cb_start(_desc_idx: u8) {}

/// phase-474 I3 -- a sample TAKEN for the subscription in `desc_idx`, emitted
/// before its callback's start: the take event, and the sample's source stamp
/// when `type_stamp` (the type's `STAMP_OFFSET`) or the slot's setting names
/// one. Opt-in per slot or image-wide (`NROS_TRACE_TAKES`).
#[cfg(feature = "trace-callbacks")]
#[inline]
fn trace_take(desc_idx: u8, raw: &[u8], type_stamp: Option<usize>) {
    super::callback_trace::take(desc_idx, raw, type_stamp);
}

#[cfg(not(feature = "trace-callbacks"))]
#[inline]
fn trace_take(_desc_idx: u8, _raw: &[u8], _type_stamp: Option<usize>) {}

/// phase-474 I3 -- whether this tick of the timer in `desc_idx` gets its
/// start/end pair (one in N, per slot or `NROS_TRACE_TIMER_EVERY`).
#[cfg(feature = "trace-callbacks")]
#[inline]
fn timer_tick_traced(desc_idx: u8) -> bool {
    super::callback_trace::timer_tick_traced(desc_idx)
}

#[cfg(not(feature = "trace-callbacks"))]
#[inline]
fn timer_tick_traced(_desc_idx: u8) -> bool {
    false
}

/// Emit callback_end(handle) immediately after a user callback returns.
#[cfg(feature = "trace-callbacks")]
#[inline]
fn trace_cb_end(desc_idx: u8) {
    super::callback_trace::end(desc_idx);
}

#[cfg(not(feature = "trace-callbacks"))]
#[inline]
fn trace_cb_end(_desc_idx: u8) {}

/// Kind of registered callback entry.
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum EntryKind {
    Subscription,
    Service,
    ServiceClient,
    Timer,
    ActionServer,
    ActionClient,
    GuardCondition,
}

/// What a registration site calls the callback it is registering.
///
/// Phase 8 (`docs/design/callback_tracing.rst`) — the payload of the
/// nros_callback_register event, resolved to a string only inside the
/// feature-gated emitter so a build without `trace-callbacks` pays nothing
/// for the two synthesised forms.
///
/// The three variants exist because the executor does NOT have a name for
/// every entry kind:
///
/// * subscriptions / services / actions carry a topic or service name;
/// * a timer carries only a period — so the period IS its identity;
/// * a guard condition, and an arena subscription attached to an already-open
///   `RmwSubscriber`, carry nothing at all — the slot index is the only
///   thing that distinguishes one from another.
///
/// Naming them here rather than at the 25 emplace sites keeps the synthesis
/// rules in one place; a new registration site that gets the name wrong is
/// then a wrong ARGUMENT, not a second convention.
// Only the feature-gated emitter READS these, so a build with the feature off
// constructs them and never matches on them. That is the intended shape (the
// call sites must read identically either way), not dead code to delete.
#[cfg_attr(not(feature = "trace-callbacks"), allow(dead_code))]
#[derive(Clone, Copy)]
pub(crate) enum TraceName<'a> {
    /// A real name the caller already has: topic, service, or action name.
    Text(&'a str),
    /// A timer, rendered `timer@<period_us>us`.
    TimerPeriod(u64),
    /// An entry with no name anywhere in the executor, rendered
    /// `<label>#<slot>` — `guard#3`, `sub#7`. The slot index is exactly what
    /// the runtime callback_start / callback_end events key on, so this
    /// label is not a placeholder: it is the identity, spelled out.
    Slot(&'static str, usize),
}

/// Metadata for a type-erased callback stored in the arena.
///
/// Each entry records where the concrete entry struct lives in the arena
/// and carries monomorphized function pointers for dispatch and cleanup.
#[derive(Clone, Copy)]
pub(crate) struct CallbackMeta {
    /// Byte offset into the arena where the concrete entry starts.
    pub(crate) offset: usize,
    /// What kind of entry this is (for `SpinOnceResult` counters).
    pub(crate) kind: EntryKind,
    /// Monomorphized dispatch: tries to receive and process one message/request.
    /// Returns `Ok(true)` if work was done, `Ok(false)` if nothing available.
    /// The `u64` parameter is `delta_us` (used by timer entries, ignored by others).
    ///
    /// The trailing `u8` is the entry's SLOT INDEX (phase 8,
    /// `docs/design/callback_tracing.rst`). Inside a leaf the only identity
    /// otherwise in scope is the address `arena_base + offset` — unique and
    /// stable, but an address, and `arena_base` is not reachable from the
    /// leaf. Threading the index in is the design's preferred fix over
    /// publishing an `offset -> index` map: it needs no side table, it cannot
    /// go stale if the arena moves, and attribution is exact at the point of
    /// use. Both producers already have the index (the drain loop, the
    /// trigger-fail timer sweep).
    ///
    /// Passed unconditionally rather than behind `feature = "trace-callbacks"`:
    /// a cfg-dependent function-pointer signature would have to be threaded
    /// through 21 leaf definitions as a macro. The cost when tracing is off is
    /// one constant register argument per dispatch that the leaf ignores.
    pub(crate) try_process: unsafe fn(*mut u8, u64, u8) -> Result<bool, TransportError>,
    /// Monomorphized readiness check: returns true if the entry has data.
    pub(crate) has_data: unsafe fn(*const u8) -> bool,
    /// Monomorphized LET pre-sample: reads data from transport into the entry's
    /// buffer without invoking the callback. No-op for non-subscription entries.
    pub(crate) pre_sample: unsafe fn(*mut u8),
    /// Per-callback invocation mode.
    pub(crate) invocation: InvocationMode,
    /// Issue 1631 — the arena bytes this entry's registration claimed,
    /// starting at `offset` (the entry plus any trailing receive region).
    ///
    /// Registration sites write [`ArenaLen::UNRECORDED`]; the ONE choke point
    /// `Executor::emplace_entry` overwrites it with the region the allocator
    /// actually handed out, so no site can state a length that disagrees with
    /// the allocation. `release_entry` gives exactly this back. Before 1631
    /// the releasing CALLER had to restate the size from the concrete type,
    /// which only works for an entry with no trailing region and a type the
    /// caller can name — neither holds for a buffered subscription or for a
    /// timer whose closure type lives in another crate.
    pub(crate) arena_len: ArenaLen,
    /// Monomorphized drop: runs destructors on the concrete entry.
    pub(crate) drop_fn: unsafe fn(*mut u8),
}

// Issue 1631 — the no-growth claim, on EVERY target this crate compiles for
// (the host test `callback_meta_len_rides_in_padding` only sees the host).
const _: () = assert!(
    core::mem::size_of::<Option<CallbackMeta>>() == 6 * core::mem::size_of::<usize>(),
    "CallbackMeta grew: `arena_len` must ride in the tail padding (issue 1631)"
);

/// Issue 1631 — the recorded length of one entry's arena region.
///
/// Sized to ride in `CallbackMeta`'s existing tail padding, so the entry
/// table (and every stated executor backing size) does not grow: the two
/// one-byte enums leave six spare bytes on a 64-bit target and two on a
/// 32-bit one. A 32-bit region of 64 KiB or more does not fit and is stored
/// as [`UNRECORDED`](Self::UNRECORDED): its release still drops the entry and
/// frees the slot, and keeps the bytes — a leak of one region, never a reuse
/// of bytes that are still in use. `callback_meta_len_rides_in_padding` pins
/// the no-growth claim.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub(crate) struct ArenaLen(ArenaLenRepr);

#[cfg(target_pointer_width = "64")]
type ArenaLenRepr = u32;
#[cfg(not(target_pointer_width = "64"))]
type ArenaLenRepr = u16;

impl ArenaLen {
    /// No length known: a release frees the slot and keeps the bytes.
    pub(crate) const UNRECORDED: Self = Self(ArenaLenRepr::MAX);

    /// The length, or [`UNRECORDED`](Self::UNRECORDED) when it does not fit.
    pub(crate) fn new(len: usize) -> Self {
        match ArenaLenRepr::try_from(len) {
            Ok(v) if v != ArenaLenRepr::MAX => Self(v),
            _ => Self::UNRECORDED,
        }
    }

    /// `None` when unrecorded.
    pub(crate) fn get(self) -> Option<usize> {
        (self != Self::UNRECORDED).then_some(self.0 as usize)
    }
}

/// phase-476 W2 — the hook that destroys a capture the arena holds.
///
/// A capture's bytes are copied into its entry's trailing region, which MOVES
/// ownership of whatever the capture owns (a C++ `std::shared_ptr`, a
/// `std::string`): the C++ side relinquishes its own copy without destroying
/// it. So the arena copy has to be destroyed exactly once, when its entry is,
/// and this `Drop` is that once — every entry that holds a capture carries one
/// of these, so it runs inside `release_entry`'s `drop_fn` (or the executor's
/// teardown), before the region is handed back.
///
/// `drop: None` is a capture with nothing to destroy, or no capture at all.
pub(crate) struct CaptureGuard {
    pub(crate) ctx: *mut core::ffi::c_void,
    pub(crate) drop: Option<unsafe extern "C" fn(*mut core::ffi::c_void)>,
}

impl Drop for CaptureGuard {
    fn drop(&mut self) {
        if let Some(drop) = self.drop {
            // SAFETY: `ctx` is the arena's copy of the capture the registration
            // supplied `drop` for, still inside the entry's region; this runs
            // once because the entry is dropped once.
            unsafe { drop(self.ctx) }
        }
    }
}

/// phase-476 W2 — a C timer callback over a capture the arena holds.
pub(crate) struct CapturedCallback {
    pub(crate) callback: unsafe extern "C" fn(*mut core::ffi::c_void),
    pub(crate) guard: CaptureGuard,
}

impl CapturedCallback {
    pub(crate) fn call(&self) {
        // SAFETY: `ctx` is the arena's copy of the caller's capture (or null
        // when there is none), alive as long as this entry; the registration's
        // contract is that `callback` reads its capture through it.
        unsafe { (self.callback)(self.guard.ctx) }
    }
}

// ============================================================================
// Concrete entry types
// ============================================================================

/// Concrete subscription entry stored in the arena (with MessageInfo).
#[repr(C)]
pub(crate) struct SubInfoEntry<M, F> {
    pub(crate) handle: session::RmwSubscriber,
    /// Issue 1860 / phase-498 W1 -- the receive slot, in the arena's trailing
    /// region and sized at registration (from the type's bound when the
    /// caller names none), as `SubBufferedRawInfoCEntry` has been since
    /// phase-408 W5b. It was an inline `[u8; RX_BUF]`, so every entry of this
    /// shape reserved the image's closure maximum whatever its own type was.
    pub(crate) buffer: TrailingBuf,
    /// Length of pre-sampled LET data (0 = not sampled).
    pub(crate) sampled_len: usize,
    pub(crate) callback: F,
    pub(crate) _phantom: PhantomData<M>,
}

/// Concrete subscription entry stored in the arena (with safety validation).
#[cfg(feature = "safety-e2e")]
#[repr(C)]
pub(crate) struct SubSafetyEntry<M, F> {
    pub(crate) handle: session::RmwSubscriber,
    /// Issue 1860 / phase-498 W1 -- the receive slot, in the arena's trailing
    /// region and sized at registration (from the type's bound when the
    /// caller names none), as `SubBufferedRawInfoCEntry` has been since
    /// phase-408 W5b. It was an inline `[u8; RX_BUF]`, so every entry of this
    /// shape reserved the image's closure maximum whatever its own type was.
    pub(crate) buffer: TrailingBuf,
    /// Length of pre-sampled LET data (0 = not sampled).
    pub(crate) sampled_len: usize,
    pub(crate) callback: F,
    pub(crate) _phantom: PhantomData<M>,
}

/// Concrete service entry stored in the arena.
#[repr(C)]
pub(crate) struct SrvEntry<Svc: RosService, F, const REQ_BUF: usize, const REPLY_BUF: usize> {
    pub(crate) handle: session::RmwServiceServer,
    pub(crate) req_buffer: [u8; REQ_BUF],
    pub(crate) reply_buffer: [u8; REPLY_BUF],
    pub(crate) callback: F,
    pub(crate) _phantom: PhantomData<Svc>,
}

/// What a periodic timer does with periods it missed while its executor
/// was blocked (issue #505).
///
/// A tier that loses the CPU for several periods — a long-running
/// callback, or preemption by a higher-priority band — resumes with
/// accumulated elapsed time worth more than one period.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum TimerOverrunPolicy {
    /// Coalesce the backlog: fire ONCE and drop the missed periods,
    /// counting them in `overruns`. Phase is preserved (the sub-period
    /// remainder carries over), so activations stay aligned to the
    /// original cadence grid instead of drifting by each stall.
    ///
    /// The default, matching rclcpp (which advances the next call time
    /// past "now") and Zephyr's `k_timer` (which coalesces expiries and
    /// reports a count).
    #[default]
    Skip,
    /// Replay every missed period, one activation per `try_process`
    /// pass, until the backlog drains. Correct for timers that count or
    /// accumulate, where each activation is a unit of work that must
    /// not be lost. A replay burst runs back-to-back at dispatch speed,
    /// NOT at the declared cadence — control loops usually want `Skip`.
    CatchUp,
}

/// Concrete timer entry stored in the arena.
///
/// The first fields (up to `callback`) share layout with [`TimerHeader`],
/// enabling type-erased access to timer state (cancel, reset, period query).
#[repr(C)]
pub(crate) struct TimerEntry<F> {
    pub(crate) period_us: u64,
    pub(crate) elapsed_us: u64,
    /// Periods dropped by [`TimerOverrunPolicy::Skip`] (issue #505).
    /// Saturating; never cleared by the dispatcher.
    pub(crate) overruns: u32,
    /// `overruns` as of the last `timer-overrun-runtime` monitor check,
    /// so the rule reports newly dropped activations rather than the
    /// running total. Lives here rather than in a table parallel to
    /// `entries`, whose capacity is a runtime slice length.
    pub(crate) overruns_reported: u32,
    pub(crate) oneshot: bool,
    pub(crate) fired: bool,
    pub(crate) cancelled: bool,
    pub(crate) overrun_policy: TimerOverrunPolicy,
    /// Which clock advances `elapsed_us` (phase-425 W4). `Steady` consumes the
    /// executor's spin delta; the others read their clock in `try_process`.
    pub(crate) clock_source: TimerClockSource,
    /// Last reading of `clock_source`, in nanoseconds. Meaningless — and never
    /// read — while `clock_source` is `Steady`.
    pub(crate) last_clock_ns: i64,
    pub(crate) callback: F,
}

/// Type-erased header for timer entries.
///
/// Shares layout with the initial fields of `TimerEntry<F>` (both `#[repr(C)]`),
/// so a `*mut TimerHeader` can safely read/write the timer state fields
/// regardless of the concrete closure type `F`.
#[repr(C)]
pub(crate) struct TimerHeader {
    pub(crate) period_us: u64,
    pub(crate) elapsed_us: u64,
    pub(crate) overruns: u32,
    pub(crate) overruns_reported: u32,
    pub(crate) oneshot: bool,
    pub(crate) fired: bool,
    pub(crate) cancelled: bool,
    pub(crate) overrun_policy: TimerOverrunPolicy,
    pub(crate) clock_source: TimerClockSource,
    pub(crate) last_clock_ns: i64,
}

/// Concrete action server entry stored in the arena.
#[repr(C)]
pub(crate) struct ActionServerArenaEntry<
    A: RosAction,
    GoalF,
    CancelF,
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
    const MAX_GOALS: usize,
> {
    pub(crate) server: ActionServer<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>,
    pub(crate) goal_callback: GoalF,
    pub(crate) cancel_callback: CancelF,
}

/// Concrete action server entry for raw (untyped) callbacks.
///
/// Uses [`ActionServerCore`] directly (no typed `ActionServer<A>` wrapper).
#[repr(C)]
pub(crate) struct ActionServerRawArenaEntry<
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
    const MAX_GOALS: usize,
> {
    pub(crate) core: ActionServerCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>,
    pub(crate) goal_callback: RawGoalCallback,
    pub(crate) cancel_callback: RawCancelCallback,
    /// Optional hook fired after the accept reply has been sent. Used by the
    /// C API so user-supplied long-running `accepted_callback`s run *after*
    /// the client has observed the accept instead of blocking the reply.
    pub(crate) accepted_callback: Option<RawAcceptedCallback>,
    pub(crate) context: *mut core::ffi::c_void,
}

/// Concrete action client entry for raw (untyped) async callbacks.
///
/// Contains the `ActionClientCore` plus callback function pointers for
/// goal response, feedback, and result. The executor polls the core's
/// non-blocking methods during `spin_once` and invokes the callbacks.
#[repr(C)]
pub(crate) struct ActionClientRawArenaEntry<
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
> {
    pub(crate) core: ActionClientCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>,
    pub(crate) goal_response_callback: Option<RawGoalResponseCallback>,
    pub(crate) feedback_callback: Option<RawFeedbackCallback>,
    pub(crate) result_callback: Option<RawResultCallback>,
    /// Issue 2030 — `Some` makes the dispatcher take cancel replies (freeing
    /// their request slots) and hand each to this callback; see
    /// [`RawCancelResponseCallback`].
    pub(crate) cancel_response_callback: Option<RawCancelResponseCallback>,
    pub(crate) context: *mut core::ffi::c_void,
}

/// Concrete service-client entry for raw (untyped) async polling.
///
/// Holds the `RmwServiceClient` plus a single-shot reply buffer and a
/// callback fn pointer. The executor dispatches via
/// `service_client_raw_try_process` which checks `reply_ready` (set by
/// the transport waker) before calling `take_response_raw`. This
/// avoids busy-polling `get_check` on every spin tick.
///
/// Single in-flight request per entry: a second `send_request` while
/// `pending` is still `true` is the user's responsibility to avoid (the
/// C wrapper checks at the call site).
///
/// Issue 2009 — `sequences` holds the ids that logical call went out under
/// (one per re-send), and a reply is dispatched only if it names one of them.
/// Every sender records its id through [`ServiceClientRawArenaEntry::record_send`].
#[repr(C)]
pub struct ServiceClientRawArenaEntry<const REPLY_BUF: usize> {
    pub handle: session::RmwServiceClient,
    pub reply_buffer: [u8; REPLY_BUF],
    pub pending: bool,
    /// Set by the transport waker when a reply arrives for this slot.
    /// Checked by `try_process` to avoid blind polling.
    pub reply_ready: core::sync::atomic::AtomicBool,
    pub callback: Option<RawResponseCallback>,
    /// Issue 2136 — when set, dispatched IN PLACE of `callback`, with the id
    /// of the request the reply answers (rclc's
    /// `rclc_executor_add_client_with_request_id`). Same `context`.
    pub request_id_callback: Option<RawResponseRequestIdCallback>,
    pub context: *mut core::ffi::c_void,
    /// Issue 2009 — the ids of the call in flight; see the type docs.
    pub sequences: ClientSequences,
}

/// Issue 2009 — the request ids of an executor-owned client's one logical
/// call: every re-send generation of it (see
/// [`MAX_CLIENT_RESEND_GENERATIONS`](crate::limits::MAX_CLIENT_RESEND_GENERATIONS)).
pub type ClientSequences =
    nros_rmw::RequestCorrelator<(), { crate::limits::MAX_CLIENT_RESEND_GENERATIONS }>;

impl<const REPLY_BUF: usize> ServiceClientRawArenaEntry<REPLY_BUF> {
    /// Record a request this entry just sent. A FRESH call (`resend` false)
    /// first forgets the ids of any earlier call, so that call's late reply is
    /// dropped instead of being dispatched as this one's; a re-send of the
    /// same call adds its generation beside the others.
    pub fn record_send(&mut self, seq: i64, resend: bool) {
        record_client_send(&mut self.sequences, seq, resend);
    }

    /// Take the reply to the call in flight, dropping any reply that names
    /// another request. On a match every generation of the call is retired,
    /// so the replies the other generations still draw are dropped too.
    pub fn take_reply(&mut self, buf: &mut [u8]) -> Result<Option<usize>, TransportError> {
        Ok(take_client_reply(&mut self.handle, buf, &mut self.sequences)?.map(|(len, _)| len))
    }
}

/// Issue 2009 — [`ServiceClientRawArenaEntry::record_send`]'s rule, shared with
/// the typed callback entry's header.
pub(crate) fn record_client_send(seqs: &mut ClientSequences, seq: i64, resend: bool) {
    if !resend {
        seqs.clear();
    } else if seq == nros_rmw::SEQUENCE_ID_UNKNOWN && seqs.contains(seq) {
        // An id-less backend's re-sends are one request: nothing to add.
        return;
    }
    seqs.insert_evicting(seq, ());
}

/// Issue 2009 — [`ServiceClientRawArenaEntry::take_reply`]'s rule.
/// Returns the reply's length and the id it was matched on.
pub(crate) fn take_client_reply(
    handle: &mut session::RmwServiceClient,
    buf: &mut [u8],
    seqs: &mut ClientSequences,
) -> Result<Option<(usize, i64)>, TransportError> {
    match nros_rmw::take_correlated(handle, buf, seqs)? {
        Some((len, seq, ())) => {
            // The remaining generations answer a call that is now complete.
            seqs.clear();
            Ok(Some((len, seq)))
        }
        None => Ok(None),
    }
}

/// Concrete service entry for raw (untyped) callbacks.
///
/// Issue 2081 — `C` is the callback's shape: [`RawServiceCallback`] (the
/// request bytes alone) or [`RawServiceRequestIdCallback`] (plus the request's
/// id). Both are one function pointer, so the entry's size does not depend on
/// which.
#[repr(C)]
pub(crate) struct SrvRawEntry<
    const REQ_BUF: usize,
    const REPLY_BUF: usize,
    C: RawServiceDispatch = RawServiceCallback,
> {
    pub(crate) handle: session::RmwServiceServer,
    pub(crate) req_buffer: [u8; REQ_BUF],
    pub(crate) reply_buffer: [u8; REPLY_BUF],
    pub(crate) callback: C,
    pub(crate) context: *mut core::ffi::c_void,
    /// phase-488 — the arena's copy of a C++ handler's capture (a `std::bind`
    /// over `this`, a capturing lambda), destroyed with the entry. `drop: None`
    /// when the context is not a capture, which is every C caller.
    pub(crate) capture: CaptureGuard,
}

/// Concrete guard condition entry stored in the arena.
#[repr(C)]
pub(crate) struct GuardConditionEntry<F> {
    pub(crate) flag: portable_atomic::AtomicBool,
    pub(crate) callback: F,
}

// ============================================================================
// QoS-driven buffered subscription entries (Phase 73)
// ============================================================================

/// Buffer strategy selected by QoS depth at subscription registration time.
///
/// The buffer data lives in a trailing region immediately after the
/// `SubBufferedEntry` struct in the arena.
pub(crate) enum BufferStrategy {
    /// `KEEP_LAST(1)`: 3 slots, latest-value semantics, writer never blocks.
    Triple(TripleBuffer),
    /// `KEEP_LAST(N)` where N > 1: N+1 slots, FIFO ordering, bounded drops.
    Ring(SpscRing),
}

impl BufferStrategy {
    /// Check if new data is available.
    pub(crate) fn has_data(&self) -> bool {
        match self {
            BufferStrategy::Triple(tb) => tb.has_data(),
            BufferStrategy::Ring(ring) => ring.has_data(),
        }
    }
}

/// The QoS depth an action client's FEEDBACK stream registers with.
///
/// Issue 1190 — was an `8u16` literal at
/// `NodeHandle::create_action_client_with_callbacks_sized`'s one call site.
/// Named because `nros-node/build.rs` restates it to price the action-client
/// arena slot, and a literal is a restatement nothing can be held to; see
/// [`arena_model_tests`].
pub const DEFAULT_ACTION_FEEDBACK_DEPTH: u16 = 8;

/// Compute the number of buffer slots and trailing region size for a given
/// QoS depth and per-slot buffer size.
///
/// Returns `(slot_count, trailing_bytes)`.
pub(crate) fn buffered_region_size(depth: u32, slot_size: usize) -> (usize, usize) {
    if depth <= 1 {
        // Triple buffer: 3 fixed slots
        (
            TripleBuffer::SLOT_COUNT,
            TripleBuffer::SLOT_COUNT * slot_size,
        )
    } else {
        let d = depth as usize;
        (SpscRing::slot_count(d), SpscRing::region_size(d, slot_size))
    }
}

/// phase-408 W5b — a SINGLE flat payload slot living in the arena's trailing
/// region, sized at REGISTRATION from the caller's `rx_buffer_hint` instead of
/// baked into the entry as `[u8; RX_BUF]`.
///
/// Why this and not [`BufferStrategy`], which the plain C raw path uses: a
/// triple buffer / ring DECOUPLES the producer's slot from the consumer's, and
/// the two entries that hold a `TrailingBuf` carry PER-SAMPLE side data beside
/// the payload — the wire attachment for
/// [`SubBufferedRawInfoCEntry`], the integrity status for
/// `SubBufferedRawSafetyCEntry`. Decoupled slots cannot carry that, which is
/// the reason both were flat in the first place (see
/// [`SubBufferedRawInfoEntry`]'s note) and it has not changed. What made them
/// EXPENSIVE was the const, not the flatness: `RX_BUF` is
/// `DEFAULT_RX_BUF_SIZE` at every call site, so a subscription that knows its
/// type's bound was still charged the image-wide default. Moving the bytes out
/// to the trailing region spends the hint on the allocation while keeping one
/// sample per dispatch.
///
/// The buffer does not own its memory — the arena does, and it outlives every
/// entry in it.
#[repr(C)]
pub(crate) struct TrailingBuf {
    ptr: *mut u8,
    len: usize,
}

impl TrailingBuf {
    /// Adopt `len` bytes of arena trailing region at `ptr`, zeroing them.
    ///
    /// The zeroing is what makes [`as_mut_slice`](Self::as_mut_slice) a safe
    /// method: the executor arena is `&mut [MaybeUninit<u8>]`, so the region is
    /// UNINITIALISED until someone writes it, and handing out a `&mut [u8]`
    /// over uninit bytes is UB even if every reader stays inside the length a
    /// receive reported. The entries this replaced held an inline
    /// `[0u8; RX_BUF]` and were therefore initialised; one memset per
    /// REGISTRATION (not per sample) keeps that property rather than trading it
    /// for the allocation saving.
    ///
    /// # Safety
    /// `ptr` must point to at least `len` writable bytes that stay valid for
    /// the lifetime of this `TrailingBuf` — i.e. a region handed back by
    /// `arena_alloc_with_trailing`, which is never reused while the entry
    /// lives.
    pub(crate) unsafe fn init(ptr: *mut u8, len: usize) -> Self {
        // Safety: the caller's contract — `len` writable bytes at `ptr`.
        unsafe { core::ptr::write_bytes(ptr, 0, len) };
        Self { ptr, len }
    }

    /// The whole slot, read-only, for a callback to borrow a received prefix.
    pub(crate) fn as_slice(&self) -> &[u8] {
        // Safety: as `as_mut_slice` -- `len` initialised bytes at `ptr`.
        unsafe { core::slice::from_raw_parts(self.ptr, self.len) }
    }

    /// The whole slot, for a receive to fill.
    pub(crate) fn as_mut_slice(&mut self) -> &mut [u8] {
        // Safety: the `init` contract — `len` writable bytes at `ptr`, valid
        // for as long as `self`, and zeroed there so they are initialised `u8`.
        unsafe { core::slice::from_raw_parts_mut(self.ptr, self.len) }
    }

    /// The slot's base address, for handing to a C callback.
    pub(crate) fn as_ptr(&self) -> *const u8 {
        self.ptr
    }
}

/// Subscription entry with QoS-driven buffer strategy (Phase 73).
///
/// Unlike the legacy single-buffer pattern, this entry
/// stores a [`BufferStrategy`] that manages a trailing buffer region
/// allocated from the arena at registration time.
///
/// # Arena layout
///
/// ```text
/// [SubBufferedEntry<M, F> struct][trailing: slot_count × slot_size bytes]
///  ↑ offset                      ↑ buffer managed by BufferStrategy
/// ```
/// W3b.5 — a contracted subscriber's age hook: the endpoint's cell plus
/// the epoch clock, captured at registration. `None` = uncontracted (or
/// no epoch source / no stamp in the type) — the take path costs one
/// `Option` branch.
pub(crate) type AgeMon = (
    &'static crate::executor::monitor::SubMonitorCell,
    Option<fn() -> u64>,
);

/// Record one stamped take on an age cell: its age when an epoch source is
/// installed, else the take alone (issue 2215, `observe_unaged`).
#[inline]
fn observe_stamp(
    cell: &crate::executor::monitor::SubMonitorCell,
    epoch: Option<fn() -> u64>,
    raw: &[u8],
    off: usize,
) {
    if let Some(stamp_us) = crate::executor::monitor::peek_stamp_us(raw, off) {
        match epoch {
            Some(epoch) => cell.observe(stamp_us, epoch()),
            None => cell.observe_unaged(),
        }
    }
}

/// W3b.5 — peek the message stamp from the raw CDR buffer and record its
/// take-age. Compiled away per-type when `M::STAMP_OFFSET` is `None`.
#[inline]
pub(crate) fn observe_age<M: RosMessage>(raw: &[u8], mon: &Option<AgeMon>) {
    if let (Some((cell, epoch)), Some(off)) = (mon, M::STAMP_OFFSET) {
        observe_stamp(cell, *epoch, raw, off);
    }
}

/// Issue 2215 — the C/C++ callback path's age hook. That path carries no Rust
/// type, so the stamp's CDR offset comes from the registration (the C++
/// template reads it off the message type, `M::STAMP_OFFSET`, and passes it in
/// `nros_cpp_subscription_options_t::stamp_offset`). Bound after registration
/// by `Executor::bind_raw_subscription_stamp`; `None` = no age row on this
/// topic, or the type carries no stamp.
#[derive(Clone, Copy)]
pub(crate) struct RawAgeMon {
    pub(crate) cell: &'static crate::executor::monitor::SubMonitorCell,
    pub(crate) epoch: Option<fn() -> u64>,
    pub(crate) stamp_offset: u16,
}

/// Issue 2215 — record a C/C++ subscription's take on its age row.
#[inline]
pub(crate) fn observe_age_raw(raw: &[u8], mon: &Option<RawAgeMon>) {
    if let Some(m) = mon {
        observe_stamp(m.cell, m.epoch, raw, m.stamp_offset as usize);
    }
}

#[repr(C)]
pub(crate) struct SubBufferedEntry<M, F> {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: BufferStrategy,
    pub(crate) callback: F,
    /// W3b.5 — age hook for contracted endpoints.
    pub(crate) age_mon: Option<AgeMon>,
    pub(crate) _phantom: PhantomData<M>,
}

/// Drain the RMW subscriber handle into the buffer strategy.
///
/// Calls `take_serialized()` on the subscriber handle and writes received data
/// into the triple buffer's write slot or the SPSC ring's next push slot.
///
/// # Safety
/// `entry` must be a valid mutable reference to a `SubBufferedEntry`.
/// Issue 0757 — how many takes this process has dropped, for throttling.
///
/// A COUNTER, not a per-entry field, and deliberately: the arena's entry
/// structs are sized by knob at build time (`EXECUTOR_OPAQUE_U64S`), so adding
/// a field here would move every image's executor footprint to buy a log line.
/// The cost is that the report cannot name the topic — see `report_dropped_take`.
static DROPPED_TAKES: portable_atomic::AtomicU32 = portable_atomic::AtomicU32::new(0);

/// issue 0900 — has the arena-headroom advisory been emitted?
///
/// A STATIC, for the reason [`DROPPED_TAKES`] is one: an `Executor` field would
/// move `EXECUTOR_OPAQUE_U64S` and therefore every image's executor footprint,
/// to buy a diagnostic. Process-scoped rather than per-executor is also the
/// RIGHT scope here, not merely the cheap one — the number this names
/// (`NROS_EXECUTOR_ARENA_SIZE`) is a BUILD-TIME constant, identical for every
/// executor in the image, so saying it twice adds nothing.
static ARENA_ADVISORY_DONE: portable_atomic::AtomicBool = portable_atomic::AtomicBool::new(false);

/// Report an arena that is far larger than the entities registered in it.
///
/// `ARENA_SIZE` is derived by budgeting EVERY slot at the ActionClient worst
/// case (`nros-node/build.rs`) unless the image declares its entities, so an
/// image with no action client carries several times what it can use — 74,240
/// bytes against ~51 KiB for four default-QoS subscriptions. (That comparison
/// read "~16 KiB" until issue 1190: the pub/sub figure it named came from a
/// model that priced a subscription's history at three slots when the ROS
/// default is eleven.) The arena is a BUMP allocator, so
/// `arena_used` is the exact claimed total rather than a reservation, and the
/// allocator has therefore always known the right answer and never said it.
///
/// The override exists (`NROS_EXECUTOR_ARENA_SIZE`, or
/// `CONFIG_NROS_EXECUTOR_ARENA_SIZE` on Zephyr) and is already load-bearing:
/// the FreeRTOS action examples pin it to 8192 and still need a 64 KB app task
/// stack, a number someone found by hitting "Invalid mbox" and working
/// backwards. That is the shape of issues 0271/0739 — a knob nobody can
/// enumerate is a knob nobody sets — and this turns it from folklore into a
/// measurement.
///
/// **Where the arena lives is the CALLER's choice**, and THREE answers are real
/// -- phase-392 W6 surveyed every entry path and found a third this comment had
/// missed. `Executor` holds `arena: &'s mut [MaybeUninit<u8>]` -- a borrowed
/// slice, since phase-271 (issue 0110) moved the sized tables off build-time
/// consts. What is inline is `ExecutorInlineStorage::backing`, which the C FFI
/// sizes its `_opaque` from.
///
/// * **`.bss`** -- all 34 in-tree C `rclc_executor_t` objects are file-scope
///   statics; the C++ entry reaches `Node::GlobalStorageHolder`, also a static;
///   and the `alloc` convenience constructors serve from
///   `super::backing::EXECUTOR_BACKING`. Visible to `nm` and `mem-report`.
///   Measured on mr-canhubk344: DTCM tracked ARENA_SIZE one-for-one across
///   MAX_CBS 24 -> 36. Measured on the native zenoh talker: 21,560 B.
/// * **the heap** -- the `alloc` constructors' fallback (a second executor, or
///   an entry sized past the reservation, or `NROS_EXECUTOR_BACKING_U64S=0`).
///   This was the ONLY Rust answer before W6, and it has no symbol, which is
///   why the campaign's own instrument could not see its largest consumer.
/// * **a task stack** -- a stack-declared `rclc_executor_t`. Legal, and nothing
///   in the tree does it.
///
/// Do not size against one placement and assume another.
///
/// `nros_log`, never stdio: this is reached on `no_std` targets and inside
/// Zephyr `native_sim`, where a Rust `std` stdio call is FATAL (issue 0589).
#[cold]
fn report_arena_headroom(used: usize, capacity: usize) {
    // Round the suggestion up so a small later registration still fits, and
    // keep it a multiple of 1024 because that is the unit the knob is written
    // in everywhere else in the tree.
    let suggest = used.next_multiple_of(1024).max(1024);
    // BUDGET: `nros_log`'s call-site format buffer is 256 bytes by default
    // (`NROS_LOG_BUFFER_SIZE`), and overflow truncates with a `…` rather than
    // dropping the record. The first draft of this line ran ~450 bytes and was
    // cut mid-number, so the sink received "executor arena is 74240 bytes and
    // 32…" — every word of the explanation and NONE of the value to set. A
    // diagnostic that explains itself past the budget delivers exactly the
    // folklore it was written to replace. So: the actionable value FIRST, the
    // reasoning in this comment and issue 0900, and a test that fails on `…`.
    nros_log::log_info!(
        nros_log::get_logger("nros"),
        // phase-392 W6 — the clause "and the arena is INLINE ON THE TASK STACK"
        // stood here and was FALSE on every path (see the doc comment above);
        // dropping it is also what buys the budget back. Placement is the
        // caller's, so this line names the NUMBER, which is the same whichever
        // memory the backing came from.
        "arena over-provisioned: set NROS_EXECUTOR_ARENA_SIZE={suggest} \
         (Zephyr: CONFIG_ prefix). {used}/{capacity} bytes claimed at first \
         spin; later registrations need more."
    );
}

/// One-shot arena-headroom check, called on the first `spin_once`.
///
/// First spin, not registration end, because there is no "registration end" —
/// an app may register lazily, and this is why the message says
/// "at first spin" and names what it measured rather than asserting a total.
///
/// Hot-path cost is one relaxed load on a branch that is taken exactly once.
pub(crate) fn maybe_report_arena_headroom(used: usize, capacity: usize) {
    if ARENA_ADVISORY_DONE.load(portable_atomic::Ordering::Relaxed) {
        return;
    }
    // Only ever set, never cleared, so a racing second spin at worst emits the
    // line twice — cheaper than an AcqRel on every spin to prevent a duplicate
    // advisory.
    ARENA_ADVISORY_DONE.store(true, portable_atomic::Ordering::Relaxed);
    if arena_is_over_provisioned(used, capacity) {
        report_arena_headroom(used, capacity);
    }
}

/// Say WHY a registration ran out of arena, once.
///
/// `NodeError::BufferTooSmall` is the same code a dozen other paths return, so
/// on a target where a return code is all you get, exhaustion here is
/// indistinguishable from a message that did not fit a receive buffer. This
/// names the entity that failed, the numbers involved, and the knobs that
/// actually govern it.
///
/// `NROS_PUBSUB_QOS_DEPTH` comes FIRST, with the depth this image was budgeted
/// for. Since phase-412 W3b the derived arena prices every subscription at that
/// depth, so a subscription created deeper than it is the ordinary way a
/// derived arena runs out — and raising the depth re-derives the arena, where a
/// hand-set `NROS_EXECUTOR_ARENA_SIZE` goes stale the next time any other term
/// moves (the rationale #775 gave for the depth knob). The line named only the
/// size and the action-client count until then, i.e. not the knob the likeliest
/// failure needs.
///
/// The counterpart to [`report_arena_headroom`], and the reason lowering
/// `NROS_EXECUTOR_ACTION_CLIENTS` is safe to suggest: too small fails at
/// REGISTRATION rather than at link, so the failure has to say so itself.
///
/// One-shot for the same reason the advisory is — the numbers do not change
/// between registrations, and a per-registration line on an RTOS target is a
/// flood (issue 0371's shape). `nros_log`, never stdio (issue 0589), and inside
/// the 256-byte format budget that truncated the first advisory — at the widest
/// realistic values too (10-digit counts, a 32-byte entity, a 5-digit depth),
/// which
/// `executor::tests::an_exhausted_arena_decodes_to_the_knob_an_operator_must_set`
/// derives from the real line and holds under the buffer.
#[cold]
pub(crate) fn report_arena_exhausted(
    entity: &'static str,
    want: usize,
    used: usize,
    capacity: usize,
    released: usize,
    request: usize,
) {
    if ARENA_EXHAUSTED_REPORTED.swap(true, portable_atomic::Ordering::Relaxed) {
        return;
    }
    let entity = entity_label(entity);
    let depth = crate::config::arena_model::BUDGETED_QOS_DEPTH;
    nros_log::log_error!(
        nros_log::get_logger("nros"),
        "arena exhausted at {entity}: {want} B short, {used}/{capacity} used. \
         Raise NROS_PUBSUB_QOS_DEPTH (budgeted {depth}) if a sub is deeper, \
         else NROS_EXECUTOR_ARENA_SIZE or NROS_EXECUTOR_ACTION_CLIENTS \
         (Zephyr: CONFIG_*)."
    );
    // Issue 1496 / 1370's question asked of THIS arena: since entries can be
    // released, the bytes may be there in holes none of which fits. A second
    // line, only then, so the common case keeps its one line and its budget.
    if arena_refusal_is_fragmented(request, released) {
        nros_log::log_error!(
            nros_log::get_logger("nros"),
            "arena FRAGMENTED: {released} B released, no hole holds {request} B \
             (issue 1496)"
        );
    }
}

/// Issue 1496 — the arena's own exhaustion classification, the same rule
/// `zpico_alloc::Exhaustion::classify` states for the platform heap (issue
/// 1370): the released bytes would hold the request, so it was the SHAPE that
/// refused it, not the size. (Released regions are reused first-fit before the
/// bump pointer moves, so reaching exhaustion with enough released bytes means
/// no single region fitted.)
pub(crate) const fn arena_refusal_is_fragmented(request: usize, released: usize) -> bool {
    released >= request && request > 0
}

/// Issue 1036 — arena exhaustion reaches the platform's FATAL HOOK where the
/// image asked for it (`NROS_ARENA_EXHAUSTION_IS_FATAL`, Zephyr
/// `CONFIG_NROS_ARENA_EXHAUSTION_IS_FATAL`, default ON with the boot report),
/// the arena's twin of `CONFIG_NROS_HEAP_EXHAUSTION_IS_FATAL` (issue 1425).
///
/// The log line goes nowhere on a board with no console, and a returned
/// `BufferTooSmall` is handled only by code written to handle it — so without
/// this, the observable outcome of an exhausted arena on the island is an image
/// that keeps running with an entity missing. The record is written BEFORE
/// this (`arena_exhausted`), so it survives the halt.
///
/// In a unit-test build the hook is a latch the test reads instead of a call
/// into a platform the test does not link.
#[cold]
pub(crate) fn arena_exhaustion_is_fatal() {
    #[cfg(all(nros_arena_exhaustion_fatal, not(test)))]
    {
        unsafe extern "C" {
            fn nros_platform_panic(msg: *const u8, len: usize) -> !;
        }
        const MSG: &[u8] = b"executor arena exhausted (see the boot report's failed_alloc_size)";
        // SAFETY: a static byte string and its length; the platform contract
        // takes any bytes and never returns.
        unsafe { nros_platform_panic(MSG.as_ptr(), MSG.len()) }
    }
    #[cfg(all(nros_arena_exhaustion_fatal, test))]
    ARENA_FATAL_HOOK_REACHED.store(true, portable_atomic::Ordering::Relaxed);
}

/// Test-build stand-in for the fatal hook — see [`arena_exhaustion_is_fatal`].
#[cfg(all(nros_arena_exhaustion_fatal, test))]
pub(crate) static ARENA_FATAL_HOOK_REACHED: portable_atomic::AtomicBool =
    portable_atomic::AtomicBool::new(false);

/// The registering entity, short enough for the log budget.
///
/// From `core::any::type_name`, whose full spelling carries every generic
/// argument and runs past 150 bytes. The generics go, and so does all but the
/// last TWO path segments: most arena entries are a module's `Entry`, so the
/// last segment alone would name nothing (`sub_buffered::Entry`, not `Entry`).
/// Capped at 32 bytes; type names are ASCII, so the cut is on a char boundary.
pub(crate) fn entity_label(full: &'static str) -> &'static str {
    let path = full.split('<').next().unwrap_or(full);
    let label = match path.rmatch_indices("::").nth(1) {
        Some((i, _)) => &path[i + 2..],
        None => path,
    };
    let cut = label.len().min(32);
    label.get(..cut).unwrap_or(label)
}

#[cfg(test)]
mod entity_label_tests {
    use super::entity_label;

    mod sub_buffered {
        pub struct Entry<T>(pub T);
    }

    /// The shape nearly every arena entry has: a module's `Entry`, generic.
    /// The last segment alone is `Entry` for all of them, which names nothing.
    #[test]
    fn a_module_entry_keeps_its_module_and_drops_its_generics() {
        let full = core::any::type_name::<sub_buffered::Entry<[u8; 1024]>>();
        assert!(full.contains('<'), "precondition: a generic name: {full}");
        assert_eq!(entity_label(full), "sub_buffered::Entry");
    }

    #[test]
    fn a_primitive_is_itself() {
        assert_eq!(entity_label(core::any::type_name::<u8>()), "u8");
    }

    #[test]
    fn a_long_label_is_capped_for_the_log_budget() {
        let label = entity_label("a::a_module_name_well_past_the_cap::AnEntryTypeName<X>");
        assert_eq!(label.len(), 32, "{label}");
        assert!(label.starts_with("a_module_name_well_past_the_cap"));
    }
}

/// One-shot latch for [`report_arena_exhausted`]. A static for the reason
/// [`DROPPED_TAKES`] is one.
static ARENA_EXHAUSTED_REPORTED: portable_atomic::AtomicBool =
    portable_atomic::AtomicBool::new(false);

/// Is this arena grossly larger than what registered in it?
///
/// Half is the threshold because the derivation's own over-provision at the
/// defaults is larger than that (74,240 bytes budgeted against ~51 KiB for four
/// default-QoS subscriptions, and against a few hundred bytes for a timer-only
/// image), so "under half" is unambiguous rather than a rounding artifact, and
/// an image that genuinely uses most of its arena stays quiet. Issue 1190 moved
/// the pub/sub half of that comparison from ~16 KiB to ~51 KiB, which
/// deliberately takes a four-subscription image OUT of the advisory: it is not
/// over-provisioned, it was mis-priced.
///
/// A zero capacity is NOT over-provisioned: that is the sentinel bug issue 0460
/// produced on Zephyr (a literal `0` forwarded instead of the derivation), and
/// it fails loudly on the first registration. Calling it over-provisioned would
/// bury a fatal misconfiguration under an advisory about wasted space.
///
/// Separated from [`maybe_report_arena_headroom`] so it is testable: the
/// reporter is one-shot on a process-scoped flag, so a test that called it
/// twice would silently check nothing the second time.
pub(crate) const fn arena_is_over_provisioned(used: usize, capacity: usize) -> bool {
    capacity != 0 && used.saturating_mul(2) <= capacity
}

// ============================================================================
// Issue 1190 — the arena model is an UPPER BOUND, asserted where the sizes are
// ============================================================================
//
// `nros-node/build.rs` sums a per-kind cost model into `ARENA_SIZE` and says of
// itself that subscription and service entries are "strictly smaller" than the
// action-client slot it budgets at. That sentence was false for four phases:
// the pub/sub term charged `3 * rx_buf + 512`, which is a `TripleBuffer` (the
// `depth <= 1` case) plus an entry allowance smaller than any entry this crate
// defines, while an ordinary `KEEP_LAST(10)` subscription claims
// `size_of(entry) + (depth + 1) * (slot + 8)`. Measured live: 12,144 bytes into
// an 8,192-byte arena, dying at `Node::register` with `BufferTooSmall`.
//
// A build script cannot check this. It has no `size_of`, and the dominant term
// is the BACKEND's handle — 584 bytes of `CffiSubscription`, most of it two
// 256-byte name buffers — which is a fact of the linked backend, not of the
// build script's environment. So the model is emitted (`config::arena_model`)
// and bounded HERE, per image, per backend, at compile time.
//
// Gated on `rmw-cffi` because that is the configuration a real image is built
// in. The unit-test configuration substitutes `MockSubscriber`, whose 8-deep
// queue of 256-byte canned takes makes it ~2.1 KiB — a test fixture, not a
// handle any image links, and sizing the model to cover it would spend 1.4 KiB
// of every slot on a mock. The depth half of the model is backend-independent
// and is checked in the unit tests below, which do run in that configuration.
#[cfg(feature = "rmw-cffi")]
const _: () = {
    use crate::config::arena_model as model;

    // The buffered subscription entries. `<(), ()>` and the C entries carry no
    // user closure, so this bounds the STRUCT the crate owns plus the backend
    // handle; a caller's own callback is captured by value on top of it and is
    // the one part no static bound can reach (`PUBSUB_STRUCT` is sized to leave
    // room for the declarative runtime's, which is the largest in this tree).
    assert!(
        core::mem::size_of::<SubBufferedEntry<(), ()>>() <= model::PUBSUB_STRUCT,
        "arena model: PUBSUB_STRUCT is below size_of::<SubBufferedEntry>() for \
         this backend — the derived arena would under-size every subscription \
         (issue 1190). Raise PUBSUB_ENTRY_STRUCT in nros-node/build.rs."
    );
    assert!(
        core::mem::size_of::<SubBufferedRawEntry<()>>() <= model::PUBSUB_STRUCT,
        "arena model: PUBSUB_STRUCT is below size_of::<SubBufferedRawEntry>() \
         for this backend (issue 1190)."
    );
    assert!(
        core::mem::size_of::<SubBufferedRawCEntry>() <= model::PUBSUB_STRUCT,
        "arena model: PUBSUB_STRUCT is below size_of::<SubBufferedRawCEntry>() \
         for this backend (issue 1190)."
    );
    assert!(
        core::mem::size_of::<SubBufferedTypedCEntry>() <= model::PUBSUB_STRUCT,
        "arena model: PUBSUB_STRUCT is below \
         size_of::<SubBufferedTypedCEntry>() for this backend (issue 1190)."
    );
    assert!(
        core::mem::size_of::<SubBufferedRawInfoCEntry>() <= model::PUBSUB_STRUCT,
        "arena model: PUBSUB_STRUCT is below \
         size_of::<SubBufferedRawInfoCEntry>() for this backend (issue 1190)."
    );

    // A service server has NO trailing region: both buffers are inline and it
    // reaches the arena through `arena_alloc`. So the whole entry is bounded.
    assert!(
        core::mem::size_of::<
            SrvRawEntry<
                { crate::config::DEFAULT_RX_BUF_SIZE },
                { crate::config::DEFAULT_RX_BUF_SIZE },
            >,
        >() <= model::SERVICE_ENTRY,
        "arena model: SERVICE_ENTRY is below size_of::<SrvRawEntry>() for this \
         backend (issue 1190)."
    );

    // A timer has no buffer at all.
    assert!(
        core::mem::size_of::<TimerEntry<()>>() <= model::TIMER_ENTRY,
        "arena model: TIMER_ENTRY is below size_of::<TimerEntry>() (issue 1190)."
    );
};

#[cfg(test)]
mod arena_model_tests {
    use crate::config::{DEFAULT_RX_BUF_SIZE, arena_model as model};
    use nros_rmw::QoSProfile;

    /// Issue 1190, the failure itself. The derivation's pub/sub term must cover
    /// the buffered region an ORDINARY subscription claims — the one created
    /// with no QoS argument, which is `QoSProfile::default()`, which is
    /// `rmw_qos_profile_default`, which is `KEEP_LAST(10)`.
    ///
    /// This is the assertion the old model fails: it charged
    /// `3 * DEFAULT_RX_BUF_SIZE` = 3,072 bytes, a `TripleBuffer`, where the
    /// allocator claims `11 * (1024 + 8)` = 11,352 — short by 8,280 per
    /// subscription, on top of a 280-byte shortfall in the entry allowance. An
    /// image with one declared subscription therefore derived an 8,192-byte
    /// arena for a 12,144-byte registration.
    ///
    /// phase-454 W5 / issue 1319 — measured at the slot size the derivation
    /// PRICED, not at `DEFAULT_RX_BUF_SIZE`.
    ///
    /// Those were the same number until issue 1255 narrowed the term to the
    /// image-wide subscribed class, and this assertion went on reading `RX_BUF`
    /// — which made it right about the two registration paths that take the
    /// closure buffer and wrong about the three that do not.
    /// A Cyclone image priced at its bound would have failed here for sizing
    /// CORRECTLY. `PUBSUB_SLOT_BYTES` is what the build actually charged, and
    /// `the_modelled_slot_never_exceeds_the_closure_buffer` below is the other
    /// half — the one that says the price is an upper bound on every path.
    ///
    /// Deliberately measured through `buffered_region_size`, the function the
    /// allocator itself calls, and not through a second copy of its arithmetic.
    #[test]
    fn the_modelled_pubsub_region_covers_a_default_qos_subscription() {
        // phase-412 W3b — against the depth this image BUDGETED, not against
        // ROS 2's default. They are the same number until something states
        // otherwise, and once something does, "does the model cover a DEFAULT
        // subscription" stops being the question the image is asking: an image
        // that budgets KEEP_LAST(1) is claiming its subscriptions are
        // KEEP_LAST(1), and this assertion holding at depth 10 would only mean
        // it had over-provisioned.
        //
        // What it still catches is the thing that matters — a budget that does
        // not cover its own claim.
        let (slots, region) =
            super::buffered_region_size(model::BUDGETED_QOS_DEPTH, model::PUBSUB_SLOT_BYTES);
        assert!(
            model::PUBSUB_REGION >= region,
            "the arena derivation budgets {} bytes for a subscription's \
             buffered region and the allocator claims {region} for it \
             ({slots} slots of {} at the budgeted \
             KEEP_LAST({}) ) — every image that derives its arena from a \
             declared subscription count is short by {} bytes per \
             subscription (issue 1190)",
            model::PUBSUB_REGION,
            model::PUBSUB_SLOT_BYTES,
            model::BUDGETED_QOS_DEPTH,
            region.saturating_sub(model::PUBSUB_REGION),
        );
    }

    /// **Issue 1319** — the slot the model prices may never exceed `RX_BUF`,
    /// because `RX_BUF` is what the widest registration path claims and the
    /// model is an upper bound or it is nothing.
    ///
    /// The direction that shipped the bug is the other one, and it has no
    /// TOTAL assertion available here: a slot priced BELOW `RX_BUF` is correct
    /// for a `TypedBound` registration (once ~~`c_typed_hint`~~ /
    /// ~~`rust_typed_descriptors`~~) and 1,848 bytes per subscription short for
    /// an `Unbounded` one, and which one this image
    /// takes is composed from the entry's language and the linked backend —
    /// neither of which the runtime can read out of a `cfg`. That is why the
    /// fact travels in the sizing descriptor and the check is on the DERIVATION
    /// (`nros-sizing-descriptor`'s slot table, `nros-node/build.rs`'s
    /// `row_slot_bytes`) rather than here. What stands here is the cheap half
    /// and the reproduction in `executor::tests`.
    ///
    /// Both sides go through `black_box`, for the reason `config.rs`'s
    /// `opaque` exists: they are build-emitted CONSTANTS, and clippy
    /// const-propagates the comparison and refuses the `assert!` as
    /// `assertions_on_constants` under `-D warnings`. A `const { assert!(…) }`
    /// would satisfy it and cost the message its two numbers, which are the
    /// whole diagnostic.
    #[test]
    fn the_modelled_slot_never_exceeds_the_closure_buffer() {
        fn opaque(v: usize) -> usize {
            core::hint::black_box(v)
        }
        assert!(
            opaque(model::PUBSUB_SLOT_BYTES) <= opaque(DEFAULT_RX_BUF_SIZE),
            "the arena model prices a subscription slot at {} bytes while the \
             widest registration path claims DEFAULT_RX_BUF_SIZE = \
             {DEFAULT_RX_BUF_SIZE} — a model term above every path it stands \
             for is not an upper bound, it is a second number (issue 1319)",
            model::PUBSUB_SLOT_BYTES,
        );
    }

    /// The depth the derivation restates must be the depth the runtime uses. A
    /// build script cannot read `QOS_PROFILE_DEFAULT`, so the number is
    /// retyped there; this is what stops the two drifting, which is exactly how
    /// the term came to model a depth nothing creates.
    #[test]
    fn the_modelled_qos_depth_is_the_runtime_default() {
        assert_eq!(
            model::QOS_DEPTH,
            QoSProfile::default().depth,
            "nros-node/build.rs budgets a subscription at KEEP_LAST({}) while \
             `QoSProfile::default()` is KEEP_LAST({}) (issue 1190)",
            model::QOS_DEPTH,
            QoSProfile::default().depth,
        );
    }

    /// Same, for the action client's feedback stream.
    #[test]
    fn the_modelled_action_feedback_depth_is_what_an_action_client_registers() {
        assert_eq!(
            model::ACTION_FEEDBACK_DEPTH,
            super::DEFAULT_ACTION_FEEDBACK_DEPTH,
            "nros-node/build.rs prices the action-client slot at feedback \
             depth {} while the registration uses {} (issue 1190)",
            model::ACTION_FEEDBACK_DEPTH,
            super::DEFAULT_ACTION_FEEDBACK_DEPTH,
        );
    }

    /// The action-client term is left depth-blind on purpose (see
    /// `build.rs`), and that is only defensible while it stays an upper bound.
    /// Its feedback region is real and depth-scaled even though the model does
    /// not spell it that way, so the region alone must fit inside the term with
    /// the entry struct still to pay for.
    #[test]
    fn the_modelled_action_client_entry_covers_its_feedback_region() {
        let (_slots, region) = super::buffered_region_size(
            super::DEFAULT_ACTION_FEEDBACK_DEPTH as u32,
            DEFAULT_RX_BUF_SIZE,
        );
        assert!(
            model::ACTION_CLIENT_ENTRY > region,
            "the action-client slot is budgeted at {} bytes and its feedback \
             region alone claims {region} — the margin that hid issue 1190 has \
             closed and the term now needs the same treatment the pub/sub one \
             got",
            model::ACTION_CLIENT_ENTRY,
        );
    }
}

/// Issue 0810 — every per-kind term the DECLARED model sums is held to the type
/// the raw C/C++ registration actually allocates, at compile time, with the
/// backend this image LINKS. A build script cannot see `size_of` of a backend
/// handle; this compilation can, so a backend whose handle grows past
/// `RMW_HANDLE_BOUND` — or an entry that outgrows its term — is a build error
/// naming the term rather than a `BufferTooSmall` at the first registration.
///
/// `rmw-cffi` is the only real `ConcreteSession`; the unit-test mock's handles
/// carry `Vec`s and `RefCell`s and say nothing about a target. `+ 8` is the
/// worst alignment pad a bump allocation pays in front of an entry.
#[cfg(feature = "rmw-cffi")]
const _: () = {
    use crate::config::{DEFAULT_RX_BUF_SIZE as B, arena_model as m};
    use core::mem::size_of;
    const G: usize = crate::limits::MAX_CONCURRENT_GOALS;
    const fn fits(actual: usize, term: usize) -> bool {
        actual + 8 <= term
    }
    assert!(
        size_of::<session::RmwPublisher>() <= m::RMW_HANDLE_BOUND
            && size_of::<session::RmwSubscriber>() <= m::RMW_HANDLE_BOUND
            && size_of::<session::RmwServiceServer>() <= m::RMW_HANDLE_BOUND
            && size_of::<session::RmwServiceClient>() <= m::RMW_HANDLE_BOUND,
        "issue 0810: an RMW handle of the linked backend exceeds RMW_HANDLE_BOUND in \
         nros-node/build.rs, so every declared action / service term under-prices it"
    );
    assert!(
        fits(
            size_of::<ActionServerRawArenaEntry<B, B, B, G>>(),
            m::ACTION_SERVER_ENTRY
        ),
        "issue 0810: ActionServerRawArenaEntry outgrew ACTION_SERVER_ENTRY (nros-node/build.rs)"
    );
    assert!(
        fits(
            size_of::<ActionClientRawArenaEntry<B, B, B>>(),
            m::ACTION_CLIENT_ENTRY
        ),
        "issue 0810: ActionClientRawArenaEntry outgrew ACTION_CLIENT_ENTRY (nros-node/build.rs)"
    );
    assert!(
        fits(size_of::<SrvRawEntry<B, B>>(), m::SERVICE_ENTRY),
        "issue 0810: SrvRawEntry outgrew SERVICE_ENTRY (nros-node/build.rs)"
    );
    assert!(
        fits(
            size_of::<ServiceClientRawArenaEntry<B>>(),
            m::SERVICE_CLIENT_ENTRY
        ),
        "issue 0810: ServiceClientRawArenaEntry outgrew SERVICE_CLIENT_ENTRY (nros-node/build.rs)"
    );
    assert!(
        fits(
            size_of::<GuardConditionEntry<(unsafe extern "C" fn(*mut core::ffi::c_void), usize)>>(),
            m::GUARD_CONDITION_ENTRY
        ),
        "issue 0810: GuardConditionEntry outgrew GUARD_CONDITION_ENTRY (nros-node/build.rs)"
    );
    assert!(
        fits(size_of::<SubBufferedRawCEntry>(), m::PUBSUB_STRUCT)
            && fits(size_of::<SubBufferedTypedCEntry>(), m::PUBSUB_STRUCT),
        "issue 0810: a C subscription entry outgrew PUBSUB_STRUCT (nros-node/build.rs)"
    );
};

#[cfg(test)]
mod per_kind_term_tests {
    use crate::config::arena_model as model;

    fn opaque(v: usize) -> usize {
        core::hint::black_box(v)
    }

    /// Issue 0810 — the fallback budgets an undeclared slot at
    /// `WORST_SLOT_ENTRY`, which is a worst case only while it covers every
    /// kind a slot could turn out to be.
    #[test]
    fn the_undeclared_slot_budget_covers_every_declared_kind() {
        for (kind, term) in [
            ("subscription", model::PUBSUB_ENTRY),
            ("service server", model::SERVICE_ENTRY),
            ("service client", model::SERVICE_CLIENT_ENTRY),
            ("timer", model::TIMER_ENTRY),
            ("guard condition", model::GUARD_CONDITION_ENTRY),
            ("action client", model::ACTION_CLIENT_ENTRY),
            ("action server", model::ACTION_SERVER_ENTRY),
        ] {
            assert!(
                opaque(model::WORST_SLOT_ENTRY) >= opaque(term),
                "WORST_SLOT_ENTRY {} is below the {kind} term {term}: an undeclared \
                 image's per-slot budget no longer bounds what a slot can hold",
                model::WORST_SLOT_ENTRY
            );
        }
    }

    /// Issue 0810 — the measured shape the issue was filed on, held as a
    /// direction rather than a literal: a declared action server costs LESS
    /// than the undeclared worst case (it was billed exactly that), and a
    /// declared service client costs something (it was billed nothing).
    #[test]
    fn declared_actions_and_service_clients_are_priced_on_their_own_terms() {
        assert!(opaque(model::ACTION_SERVER_ENTRY) < opaque(model::WORST_SLOT_ENTRY));
        assert!(opaque(model::ACTION_CLIENT_ENTRY) < opaque(model::WORST_SLOT_ENTRY));
        assert!(opaque(model::SERVICE_CLIENT_ENTRY) > opaque(crate::config::DEFAULT_RX_BUF_SIZE));
        assert!(opaque(model::GUARD_CONDITION_ENTRY) > 0);
    }
}

#[cfg(test)]
mod arena_headroom_tests {
    use super::arena_is_over_provisioned;
    use crate::config::{
        ARENA_ACTION_CLIENTS, ARENA_SIZE, DEFAULT_RX_BUF_SIZE, MAX_CBS, arena_model as model,
    };

    /// issue 0900 — the per-kind derivation must reproduce the OLD
    /// `max_cbs * action_client_entry + base` arithmetic byte for byte when
    /// every slot is still budgeted at ActionClient size, which is the default.
    ///
    /// This is the compatibility gate: the knob exists so an image CAN shrink
    /// its arena, not so every image silently does. A change here that moves
    /// the default is a change to every image's stack frame.
    #[test]
    fn the_default_derivation_is_unchanged() {
        if ARENA_ACTION_CLIENTS != MAX_CBS {
            // The test build set the knob; the identity below is not the claim
            // being made then. Fail rather than pass vacuously.
            panic!(
                "this test asserts the DEFAULT derivation, but \
                 NROS_EXECUTOR_ACTION_CLIENTS was set to {ARENA_ACTION_CLIENTS} \
                 against MAX_CBS {MAX_CBS}"
            );
        }
        // Issue 1190 — these six constants used to be RETYPED here, which is a
        // second copy of the model that agrees with the first until one of them
        // moves. They come from `config::arena_model` now, which is what
        // `build.rs` actually summed.
        // Issue 0810 — the fallback's per-slot budget is `WORST_SLOT_ENTRY`
        // now; `ACTION_CLIENT_ENTRY` prices a DECLARED action client.
        let want = (MAX_CBS * model::WORST_SLOT_ENTRY + model::BASE_OVERHEAD).max(model::FLOOR);
        assert_eq!(
            ARENA_SIZE, want,
            "the per-kind derivation moved the default arena; every image's \
             executor backing moves with it (issue 0900)"
        );
        // And the shipped number itself, because reading it out of the model
        // proves the SHAPE and not the size: with both sides emitted from one
        // build script, a term could move and the identity above would still
        // hold. 74,240 is what 1,524 of the 1,547 built configs in this tree
        // carry; issue 1190's fix is required to leave it exactly there,
        // because the pub/sub term it corrects multiplies
        // `MAX_CBS - ARENA_ACTION_CLIENTS`, which is zero at the defaults.
        //
        // Only when the INPUTS are the shipped ones. A test build that raised
        // `NROS_EXECUTOR_MAX_CBS` or the buffer size derives a different number
        // legitimately, and asserting a literal at it would be this file's own
        // "second copy" mistake wearing a different hat.
        if MAX_CBS == 4 && DEFAULT_RX_BUF_SIZE == 1024 {
            assert_eq!(
                ARENA_SIZE, 74_240,
                "the shipped default arena moved. That is not forbidden, but \
                 it is a change to every image built without an entity \
                 declaration and it has to be a decision, not a side effect \
                 (issues 0900, 1190)"
            );
        }
    }

    /// The advisory must actually fire for the shipped defaults — otherwise W1
    /// installed a diagnostic that is dead on the very configuration that
    /// needs it. A timer-only executor claims 32 bytes against ARENA_SIZE.
    #[test]
    fn the_shipped_default_arena_trips_the_advisory() {
        assert!(
            arena_is_over_provisioned(32, ARENA_SIZE),
            "a timer-only executor must trip the advisory at the shipped \
             defaults; ARENA_SIZE is {ARENA_SIZE}"
        );
    }

    #[test]
    fn gross_over_provision_is_reported() {
        // The measured shape: a talker's handful of entries against the
        // worst-case-derived 74,240.
        assert!(arena_is_over_provisioned(4_096, 74_240));
        assert!(arena_is_over_provisioned(0, 74_240));
    }

    #[test]
    fn a_well_sized_arena_stays_quiet() {
        assert!(!arena_is_over_provisioned(60_000, 74_240));
        // Exactly half is the boundary and IS reported; one byte more is not.
        assert!(arena_is_over_provisioned(37_120, 74_240));
        assert!(!arena_is_over_provisioned(37_121, 74_240));
    }

    #[test]
    fn a_zero_capacity_arena_is_a_fault_not_headroom() {
        // Issue 0460's sentinel bug. It fails on the first registration; an
        // advisory about wasted space would bury that.
        assert!(!arena_is_over_provisioned(0, 0));
    }

    #[test]
    fn overflow_cannot_panic_the_check() {
        // `used` is bounded by `capacity` in practice, but the doubling must
        // not be the thing that decides that.
        assert!(!arena_is_over_provisioned(usize::MAX, 74_240));
    }
}

/// Say that a take was thrown away, on the first one and every 64th after.
///
/// Issue 0757, RFC-0052 fail-loud. `take_serialized` returns `BufferTooSmall` when
/// a reassembled sample exceeds the subscription buffer, and this path used to
/// discard EVERY non-OK take. At transport level cyclone has already completed
/// and ACKed the sample by then, so the subscription looks matched and healthy
/// from every outside probe (`ros2 topic info -v`, tshark ACKNACK analysis)
/// while the application waits forever. That is how 13.4 KiB Autoware
/// trajectories were silently dropped by every Zephyr image for the whole life
/// of the lane: small degenerate samples fit the 1 KiB default, so every green
/// marker stayed green, and attribution needed a consumer-side tshark session.
///
/// **What this can and cannot say.** The buffer capacity is known here and is
/// the actionable half — it names the knob to raise
/// (`NROS_SUBSCRIPTION_BUFFER_SIZE`, or `NROS_SUBSCRIBER_BUFFER_SIZE` /
/// `ZPICO_SUBSCRIBER_LARGE_SIZE` on zenoh). The SAMPLE size is known when the
/// backend knows it: issue 1612 made a too-small `take` carry the size the
/// sample needed (`rmw_entity.h`), which reaches here as
/// `Subscription::refused_sample_len`, and issue 1632 made this line print it
/// beside the buffer (`?` when unknown — `nros_rmw::RefusedLen`'s rule, which
/// the C++ drop log shares). The topic is still not known: `SubBufferedEntry`
/// carries no name and adding one changes arena sizing.
///
/// `nros_log`, never stdio: this site is reached on `no_std` targets and inside
/// Zephyr `native_sim`, where a Rust `std` stdio call is FATAL (issue 0589).
/// `take_serialized`, with a refusal reported through [`report_dropped_take`]
/// together with the size the backend says the sample needed — the one
/// spelling of the five buffered drain sites.
fn take_reporting_drop(
    handle: &mut session::RmwSubscriber,
    slot: &mut [u8],
) -> Result<Option<usize>, TransportError> {
    let cap = slot.len();
    handle.take_serialized(slot).inspect_err(|e| {
        report_dropped_take(e, cap, handle.refused_sample_len());
    })
}

#[cold]
fn report_dropped_take(err: &TransportError, buf_len: usize, recorded: Option<usize>) {
    let n = DROPPED_TAKES.fetch_add(1, portable_atomic::Ordering::Relaxed);
    if drop_is_reported(n) {
        log_dropped_take(err, buf_len, recorded, n);
    }
}

/// First, then every 64th. A 40-participant graph must not turn one
/// misconfigured subscription into a log flood (issue 0371's shape).
///
/// Apart from [`report_dropped_take`] so a test asks it with a count of its
/// own, never through [`DROPPED_TAKES`], which every cell in the binary that
/// drops a take advances (issue 2082).
fn drop_is_reported(n: u32) -> bool {
    n == 0 || n.is_multiple_of(64)
}

/// The line for drop number `n + 1`, unconditionally.
#[cold]
fn log_dropped_take(err: &TransportError, buf_len: usize, recorded: Option<usize>, n: u32) {
    let needed = nros_rmw::RefusedLen::for_error(err, recorded);
    nros_log::log_error!(
        nros_log::get_logger("nros"),
        "subscription take DROPPED ({err:?}): {needed}-byte sample, {buf_len}-byte \
         buffer. Received and ACKed, then discarded — raise the subscription \
         buffer knob if this is BufferTooSmall. Dropped {} so far (issue 0757)",
        n + 1
    );
}

unsafe fn drain_into_buffer<M, F>(
    entry: &mut SubBufferedEntry<M, F>,
) -> Result<(), TransportError> {
    match &entry.buffer {
        BufferStrategy::Triple(tb) => {
            let slot = tb.write_slot();
            if let Some(len) = take_reporting_drop(&mut entry.handle, slot)? {
                tb.writer_publish(len);
            }
        }
        BufferStrategy::Ring(ring) => {
            while let Some(slot) = ring.try_push() {
                match take_reporting_drop(&mut entry.handle, slot)? {
                    Some(len) => ring.commit_push(len),
                    None => break,
                }
            }
        }
    }
    Ok(())
}

/// Monomorphized dispatch for buffered subscriptions.
///
/// First drains the RMW subscriber into the buffer strategy (triple buffer
/// or SPSC ring), then dispatches from the buffer to the user callback.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedEntry<M, F>`.
pub(crate) unsafe fn sub_buffered_try_process<M, F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    M: RosMessage,
    F: FnMut(&M),
{
    let entry = unsafe { &mut *(ptr as *mut SubBufferedEntry<M, F>) };

    // Phase 1: drain RMW subscriber → buffer strategy
    // Issue 0757 — let a transport error OUT, exactly as issue 0737 does for
    // the C copy below. Anything already buffered from an earlier spin is still
    // dispatched on the next call.
    unsafe { drain_into_buffer(entry)? };

    // Phase 2: dispatch from buffer → user callback
    match &entry.buffer {
        // Phase 8 — hooked. Triple-buffered: at most one invocation per
        // `try_process`, and none at all when `reader_acquire` is empty.
        // The two deserialization `?`s sit BEFORE the start hook, so a
        // malformed sample returns early without opening a span.
        BufferStrategy::Triple(tb) => match tb.reader_acquire() {
            Some((data, len)) => {
                observe_age::<M>(&data[..len], &entry.age_mon);
                let mut reader = CdrReader::new_with_header(&data[..len])
                    .map_err(|_| TransportError::DeserializationError)?;
                let msg = M::deserialize(&mut reader)
                    .map_err(|_| TransportError::DeserializationError)?;
                trace_take(desc_idx, &data[..len], <M as RosMessage>::STAMP_OFFSET);
                trace_cb_start(desc_idx);
                (entry.callback)(&msg);
                trace_cb_end(desc_idx);
                Ok(true)
            }
            None => Ok(false),
        },
        // Phase 8 — hooked INSIDE the loop: a ring drains N queued messages
        // per `try_process` and fires the callback once per message, so N
        // messages must produce N spans, not one.
        BufferStrategy::Ring(ring) => {
            let mut did_work = false;
            while let Some((data, len)) = ring.try_pop() {
                observe_age::<M>(&data[..len], &entry.age_mon);
                let mut reader = CdrReader::new_with_header(&data[..len])
                    .map_err(|_| TransportError::DeserializationError)?;
                let msg = M::deserialize(&mut reader)
                    .map_err(|_| TransportError::DeserializationError)?;
                trace_take(desc_idx, &data[..len], <M as RosMessage>::STAMP_OFFSET);
                trace_cb_start(desc_idx);
                (entry.callback)(&msg);
                trace_cb_end(desc_idx);
                ring.commit_pop();
                did_work = true;
            }
            Ok(did_work)
        }
    }
}

/// Readiness check for buffered subscriptions.
///
/// Checks the RMW subscriber handle first (new data available from transport),
/// then the buffer strategy (data already drained into triple buffer/ring).
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedEntry<M, F>`.
pub(crate) unsafe fn sub_buffered_has_data<M, F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedEntry<M, F>) };
    // Check RMW handle first (data may be in static buffer, not yet drained)
    entry.handle.has_data() || entry.buffer.has_data()
}

// ============================================================================
// In-place typed subscription (Phase 231 Wave 0.2 — RFC-0038)
// ============================================================================

/// The most samples ONE in-place dispatch takes before it returns (issue 1956).
///
/// An in-place entry reads the backend's receive ring directly, and that ring
/// is refilled by a transport task that may run at a HIGHER priority than the
/// executor. "Drain until empty" was then unbounded: under a flood the check
/// for "empty" never came true, one `try_process` never returned, and no other
/// entry on the executor (or any lower tier on the core) ran again. The
/// buffered shapes never had this — they drain what was copied in.
///
/// Why 64 and not 1: every sample that was in the ring when the dispatch began
/// should be taken in it, because one left behind waits for the NEXT spin, and
/// a spin parks until something wakes it. A sample that ARRIVES during the
/// dispatch raises that wake itself, so leaving it costs nothing. 64 covers
/// every shipped ring depth (zenoh's default is 4) with room; a ring
/// configured deeper still makes progress, a spin at a time.
pub(crate) const IN_PLACE_DRAIN_LIMIT: usize = 64;

/// Run one in-place `step` (one `process_raw_in_place`) until it reports no
/// sample or [`IN_PLACE_DRAIN_LIMIT`] samples have been taken. The one
/// spelling of the in-place drain, so no shape can fall back to "until
/// empty". Returns whether any sample was dispatched.
#[inline]
pub(crate) fn drain_in_place(
    mut step: impl FnMut() -> Result<bool, TransportError>,
) -> Result<bool, TransportError> {
    let mut did_work = false;
    for _ in 0..IN_PLACE_DRAIN_LIMIT {
        if !step()? {
            break;
        }
        did_work = true;
    }
    Ok(did_work)
}

/// In-place typed subscription entry — **no arena buffer**.
///
/// Unlike [`SubBufferedEntry`], this carries no trailing `BufferStrategy`: the
/// callback deserializes directly from the backend's borrowed receive slot via
/// [`Subscriber::process_raw_in_place`], so copy #1 (ring → arena) and the arena
/// buffer are both gone. Selected at registration when the backend advertises
/// `supports_process_in_place()`.
#[repr(C)]
pub(crate) struct SubInplaceEntry<M, F> {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) callback: F,
    /// W3b.5 — age hook for contracted endpoints.
    pub(crate) age_mon: Option<AgeMon>,
    pub(crate) _phantom: PhantomData<M>,
}

/// Monomorphized in-place dispatch for typed subscriptions.
///
/// Drains pending messages from the backend, up to [`IN_PLACE_DRAIN_LIMIT`],
/// deserializing + invoking the callback directly from each borrowed slot. Returns `Ok(true)` if any message
/// was dispatched.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubInplaceEntry<M, F>`.
pub(crate) unsafe fn sub_inplace_try_process<M, F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    M: RosMessage,
    F: FnMut(&M),
{
    let entry = unsafe { &mut *(ptr as *mut SubInplaceEntry<M, F>) };
    // Split-borrow the handle and callback (disjoint fields).
    let SubInplaceEntry {
        handle,
        callback,
        age_mon,
        ..
    } = entry;
    drain_in_place(|| {
        let mut deser_err = false;
        // Phase 8 — hooked inside the `Ok(msg)` arm of the borrow closure,
        // which the drain loop re-enters once per pending message: N
        // messages therefore produce N spans. The pair is fully contained
        // in the closure, so the `?` on `process_raw_in_place` below can
        // never fire between a start and its end. Deserialization failures
        // take the `deser_err` arms and emit nothing.
        let processed = handle.process_raw_in_place(|raw| {
            observe_age::<M>(raw, age_mon);
            match CdrReader::new_with_header(raw) {
                Ok(mut reader) => match M::deserialize(&mut reader) {
                    Ok(msg) => {
                        trace_take(desc_idx, raw, <M as RosMessage>::STAMP_OFFSET);
                        trace_cb_start(desc_idx);
                        (callback)(&msg);
                        trace_cb_end(desc_idx);
                    }
                    Err(_) => deser_err = true,
                },
                Err(_) => deser_err = true,
            }
        })?;
        if deser_err {
            return Err(TransportError::DeserializationError);
        }
        Ok(processed)
    })
}

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

    /// Issue 1956 — a ring a higher-priority task keeps refilling is never
    /// empty at the check. The drain has to come back anyway.
    #[test]
    fn a_ring_that_never_empties_still_returns() {
        let mut taken = 0usize;
        let r = drain_in_place(|| {
            taken += 1;
            Ok(true)
        });
        assert_eq!(r, Ok(true));
        assert_eq!(taken, IN_PLACE_DRAIN_LIMIT);
    }

    #[test]
    fn an_emptied_ring_stops_the_drain_early() {
        let mut left = 3usize;
        let mut calls = 0usize;
        let r = drain_in_place(|| {
            calls += 1;
            Ok(if left == 0 {
                false
            } else {
                left -= 1;
                true
            })
        });
        assert_eq!(r, Ok(true));
        assert_eq!(calls, 4, "three samples, then the empty check");
        assert_eq!(drain_in_place(|| Ok(false)), Ok(false));
    }

    #[test]
    fn a_failed_take_ends_the_drain_with_its_error() {
        let mut calls = 0usize;
        let r = drain_in_place(|| {
            calls += 1;
            if calls == 2 {
                Err(TransportError::DeserializationError)
            } else {
                Ok(true)
            }
        });
        assert_eq!(r, Err(TransportError::DeserializationError));
        assert_eq!(calls, 2);
    }
}

/// Readiness check for in-place typed subscriptions.
///
/// # Safety
/// `ptr` must point to a valid `SubInplaceEntry<M, F>`.
pub(crate) unsafe fn sub_inplace_has_data<M, F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubInplaceEntry<M, F>) };
    entry.handle.has_data()
}

// ============================================================================
// Zero-copy raw buffered subscription (Phase 73.10)
// ============================================================================

/// Buffered subscription entry for zero-copy raw callbacks.
///
/// The callback receives `&[u8]` (CDR data) borrowing directly from the
/// triple buffer's read slot or SPSC ring's pop slot. For borrowed message
/// types (e.g., `Image<'a>`), the callback calls `deserialize_view()`
/// on the data, giving the message a lifetime tied to the callback scope.
#[repr(C)]
pub(crate) struct SubBufferedRawEntry<F> {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: BufferStrategy,
    pub(crate) callback: F,
}

/// Drain helper for raw buffered entries.
///
/// Issue 0757 — the THIRD copy of this drain, and it had the same swallow as
/// the typed one above. Fixed the way issue 0737 fixed the C copy
/// (`drain_into_buffer_raw_c`): let the error OUT so `spin_once` counts it,
/// rather than inventing a second remedy for one defect.
unsafe fn drain_into_buffer_raw<F>(
    entry: &mut SubBufferedRawEntry<F>,
) -> Result<(), TransportError> {
    match &entry.buffer {
        BufferStrategy::Triple(tb) => {
            let slot = tb.write_slot();
            if let Some(len) = take_reporting_drop(&mut entry.handle, slot)? {
                tb.writer_publish(len);
            }
        }
        BufferStrategy::Ring(ring) => {
            while let Some(slot) = ring.try_push() {
                match take_reporting_drop(&mut entry.handle, slot)? {
                    Some(len) => ring.commit_push(len),
                    None => break,
                }
            }
        }
    }
    Ok(())
}

/// Dispatch for zero-copy raw buffered subscriptions.
///
/// Drains the RMW handle into the buffer, then passes the raw CDR slice
/// to the callback. The callback borrows from the buffer slot — no copy.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawEntry<F>`.
pub(crate) unsafe fn sub_buffered_raw_try_process<F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    F: FnMut(&[u8]),
{
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawEntry<F>) };

    // Issue 0757 — see `drain_into_buffer`; the error reaches `spin_once`'s
    // `subscription_errors` instead of vanishing.
    unsafe { drain_into_buffer_raw(entry)? };

    match &entry.buffer {
        // Phase 8 — hooked. Triple-buffered: one invocation per
        // `try_process` at most, none when `reader_acquire` is empty.
        BufferStrategy::Triple(tb) => match tb.reader_acquire() {
            Some((data, len)) => {
                trace_take(desc_idx, &data[..len], None);
                trace_cb_start(desc_idx);
                (entry.callback)(&data[..len]);
                trace_cb_end(desc_idx);
                Ok(true)
            }
            None => Ok(false),
        },
        // Phase 8 — hooked INSIDE the loop: one span per drained message,
        // not one per `try_process`.
        BufferStrategy::Ring(ring) => {
            let mut did_work = false;
            while let Some((data, len)) = ring.try_pop() {
                trace_take(desc_idx, &data[..len], None);
                trace_cb_start(desc_idx);
                (entry.callback)(&data[..len]);
                trace_cb_end(desc_idx);
                ring.commit_pop();
                did_work = true;
            }
            Ok(did_work)
        }
    }
}

/// Readiness check for raw buffered subscriptions.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawEntry<F>`.
pub(crate) unsafe fn sub_buffered_raw_has_data<F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawEntry<F>) };
    entry.handle.has_data() || entry.buffer.has_data()
}

// ============================================================================
// Borrowed (zero-copy) buffered subscription (Phase 229.6, issue 0007)
// ============================================================================

/// Buffered subscription entry for borrowed (zero-copy) message callbacks.
///
/// The callback receives `&B::View<'a>` — a lifetime-carrying message whose
/// unbounded sequence/string fields borrow directly from the triple buffer's
/// read slot (no arena copy, no `heapless::Vec` copy). The view is materialised
/// per dispatch via [`DeserializeView`] and dropped before the slot is
/// released, so the borrow never outlives the buffer.
///
/// **Triple-buffer only.** A borrowed view must reference exactly one
/// well-defined slot for the duration of the callback; an SPSC ring (depth > 1)
/// holds several samples in flight with no single such slot. Registration
/// rejects `qos.depth > 1` for borrowed subscriptions, so `buffer` is always
/// [`BufferStrategy::Triple`] here.
#[repr(C)]
pub(crate) struct SubBufferedViewEntry<B, F> {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: BufferStrategy,
    pub(crate) callback: F,
    pub(crate) _phantom: PhantomData<B>,
}

/// Dispatch for borrowed (zero-copy) buffered subscriptions.
///
/// Drains the RMW handle into the triple buffer, then materialises a borrowed
/// `B::View<'_>` over the read slot and hands it to the callback. The view
/// borrows the slot; it is dropped at the end of the callback, before the next
/// dispatch can publish over the slot.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedViewEntry<B, F>`.
pub(crate) unsafe fn sub_buffered_view_try_process<B, F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    B: ViewableMessage,
    F: for<'a> FnMut(&B::View<'a>),
{
    let entry = unsafe { &mut *(ptr as *mut SubBufferedViewEntry<B, F>) };

    // Borrowed subscriptions are triple-buffer only (enforced at registration).
    let tb = match &entry.buffer {
        BufferStrategy::Triple(tb) => tb,
        // Unreachable: registration rejects depth > 1. Treat as no work.
        BufferStrategy::Ring(_) => return Ok(false),
    };

    // Phase 1: drain RMW subscriber → triple buffer write slot.
    //
    // Issue 0757 — the FOURTH copy of this drain (the borrowed/zero-copy path)
    // and it swallowed non-OK takes like the others. Same remedy as issue 0737's
    // C copy: report the actionable size, then let the error out.
    {
        let slot = tb.write_slot();
        if let Some(len) = take_reporting_drop(&mut entry.handle, slot)? {
            tb.writer_publish(len);
        }
    }

    // Phase 2: borrow the read slot and deserialize a view over it (no copy).
    //
    // Phase 8 — hooked. Exactly ONE site: the `BufferStrategy::Ring(_)` arm
    // above returns `Ok(false)` without ever reaching a callback (borrowed
    // subscriptions are triple-buffer only, enforced at registration), so it
    // gets no pair. Both `?`s land before the start hook.
    match tb.reader_acquire() {
        Some((data, len)) => {
            let mut reader = CdrReader::new_with_header(&data[..len])
                .map_err(|_| TransportError::DeserializationError)?;
            let msg = <B::View<'_> as DeserializeView>::deserialize_view(&mut reader)
                .map_err(|_| TransportError::DeserializationError)?;
            trace_take(desc_idx, &data[..len], None);
            trace_cb_start(desc_idx);
            (entry.callback)(&msg);
            trace_cb_end(desc_idx);
            Ok(true)
        }
        None => Ok(false),
    }
}

/// Readiness check for borrowed buffered subscriptions.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedViewEntry<B, F>`.
pub(crate) unsafe fn sub_buffered_view_has_data<B, F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedViewEntry<B, F>) };
    entry.handle.has_data() || entry.buffer.has_data()
}

// ============================================================================
// Raw buffered subscription with attachment / MessageInfo (Phase 189.M1)
// ============================================================================

/// Staging cap for a raw subscription's wire attachment (bridge_origin
/// tags and similar are small). Attachment bytes longer than this are
/// truncated by the backend's `take_serialized_with_attachment`.
pub(crate) const RAW_INFO_ATT_CAP: usize = 256;

/// Raw buffered subscription entry that surfaces the sample's wire
/// attachment as a [`RawMessageInfo`] to the callback
/// (`FnMut(&[u8], &RawMessageInfo)`).
///
/// Unlike [`SubBufferedRawEntry`] (Triple/Ring `BufferStrategy`), this
/// uses a flat inline payload buffer + a flat attachment buffer so the
/// attachment travels with its message — the decoupled producer/consumer
/// slots of a triple/ring buffer cannot carry per-message side data.
/// One sample per dispatch (mirrors [`SubInfoEntry`]).
#[repr(C)]
pub(crate) struct SubBufferedRawInfoEntry<F> {
    pub(crate) handle: session::RmwSubscriber,
    /// Issue 1860 / phase-498 W1 -- the receive slot, in the arena's trailing
    /// region and sized at registration (from the type's bound when the
    /// caller names none), as `SubBufferedRawInfoCEntry` has been since
    /// phase-408 W5b. It was an inline `[u8; RX_BUF]`, so every entry of this
    /// shape reserved the image's closure maximum whatever its own type was.
    pub(crate) buffer: TrailingBuf,
    pub(crate) att: [u8; RAW_INFO_ATT_CAP],
    pub(crate) callback: F,
}

/// Dispatch for raw buffered subscriptions with attachment.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawInfoEntry<F>`.
pub(crate) unsafe fn sub_buffered_raw_info_try_process<F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    F: FnMut(&[u8], &RawMessageInfo),
{
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawInfoEntry<F>) };
    match entry
        .handle
        .take_serialized_with_attachment(entry.buffer.as_mut_slice(), &mut entry.att)
    {
        // Phase 8 — hooked. One sample per dispatch (flat inline buffer, no
        // ring), so exactly one pair. `Ok(None)` and `Err(_)` never reach a
        // callback and stay bare.
        Ok(Some((len, att_len))) => {
            let info = RawMessageInfo::new(&entry.att[..att_len]);
            trace_take(desc_idx, &entry.buffer.as_slice()[..len], None);
            trace_cb_start(desc_idx);
            (entry.callback)(&entry.buffer.as_slice()[..len], &info);
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Readiness check for raw buffered subscriptions with attachment.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawInfoEntry<F>`.
pub(crate) unsafe fn sub_buffered_raw_info_has_data<F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawInfoEntry<F>) };
    entry.handle.has_data()
}

/// Issue 2036 — raw buffered subscription whose closure receives the sample's
/// [`MessageInfo`] (`take_serialized_with_info`): the timestamps, sequence
/// numbers and publisher gid a backend reports, or `None` from one that
/// reports none. The raw twin of [`SubInfoEntry`], for a caller that
/// deserializes itself — the C++ `(const M&, const rclcpp::MessageInfo&)`
/// subscription callback. [`SubBufferedRawInfoEntry`] carries the ATTACHMENT
/// bytes instead, and its `RawMessageInfo` metadata is defaulted.
#[repr(C)]
pub(crate) struct SubBufferedRawMsgInfoEntry<F> {
    pub(crate) handle: session::RmwSubscriber,
    /// The receive slot, in the arena's trailing region, sized at registration.
    pub(crate) buffer: TrailingBuf,
    pub(crate) callback: F,
}

/// Dispatch for [`SubBufferedRawMsgInfoEntry`]: one sample per call.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawMsgInfoEntry<F>`.
pub(crate) unsafe fn sub_buffered_raw_msg_info_try_process<F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    F: FnMut(&[u8], Option<&MessageInfo>),
{
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawMsgInfoEntry<F>) };
    match entry
        .handle
        .take_serialized_with_info(entry.buffer.as_mut_slice())
    {
        // Phase 8 — hooked, one pair per sample, as the other raw entries.
        Ok(Some((len, info))) => {
            trace_take(desc_idx, &entry.buffer.as_slice()[..len], None);
            trace_cb_start(desc_idx);
            (entry.callback)(&entry.buffer.as_slice()[..len], info.as_ref());
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Readiness check for [`SubBufferedRawMsgInfoEntry`].
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawMsgInfoEntry<F>`.
pub(crate) unsafe fn sub_buffered_raw_msg_info_has_data<F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawMsgInfoEntry<F>) };
    entry.handle.has_data()
}

/// C-style (fn-ptr + context) raw buffered subscription with attachment
/// (Phase 189.M3.4 — the C analog of [`SubBufferedRawInfoEntry`]). Flat
/// payload + attachment buffers, one sample per dispatch.
///
/// phase-408 W5b — the payload slot is a [`TrailingBuf`] in the arena's
/// trailing region rather than an inline `[u8; RX_BUF]`, so the registering
/// caller's `rx_buffer_hint` sizes the ALLOCATION and not just the backend's
/// payload size class. The attachment stays inline and fixed
/// ([`RAW_INFO_ATT_CAP`]): it is small, and it is the thing that has to travel
/// with its message.
#[repr(C)]
pub(crate) struct SubBufferedRawInfoCEntry {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: TrailingBuf,
    pub(crate) att: [u8; RAW_INFO_ATT_CAP],
    pub(crate) callback: RawSubscriptionInfoCallback,
    pub(crate) context: *mut core::ffi::c_void,
}

/// Dispatch for the C-style raw buffered subscription with attachment.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawInfoCEntry`.
pub(crate) unsafe fn sub_buffered_raw_info_c_try_process(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawInfoCEntry) };
    let payload = entry.buffer.as_mut_slice();
    match entry
        .handle
        .take_serialized_with_attachment(payload, &mut entry.att)
    {
        // Phase 8 — hooked. One sample per dispatch; the pair brackets the
        // whole `unsafe` FFI call, which is the user callback itself.
        Ok(Some((len, att_len))) => {
            trace_take(
                desc_idx,
                unsafe { core::slice::from_raw_parts(entry.buffer.as_ptr(), len) },
                None,
            );
            trace_cb_start(desc_idx);
            unsafe {
                (entry.callback)(
                    entry.buffer.as_ptr(),
                    len,
                    entry.att.as_ptr(),
                    att_len,
                    entry.context,
                )
            };
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Readiness check for the C-style raw buffered subscription with attachment.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawInfoCEntry`.
pub(crate) unsafe fn sub_buffered_raw_info_c_has_data(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawInfoCEntry) };
    entry.handle.has_data()
}

/// Phase 250 (Wave 2) — generic (type-erased) raw buffered subscription that
/// surfaces E2E [`IntegrityStatus`](nros_rmw::IntegrityStatus) (CRC + sequence
/// gap/dup) alongside the raw CDR bytes (`FnMut(&[u8], &IntegrityStatus)`).
///
/// The type-erased analog of [`SubSafetyEntry`]: the validator lives in the
/// `RmwSubscriber` and `take_validated` produces the status, so no typed
/// `M` is needed (the declarative `Node` path is generic). Flat inline payload
/// buffer; one sample per dispatch.
#[cfg(feature = "safety-e2e")]
#[repr(C)]
pub(crate) struct SubBufferedRawSafetyEntry<F> {
    pub(crate) handle: session::RmwSubscriber,
    /// Issue 1860 / phase-498 W1 -- the receive slot, in the arena's trailing
    /// region and sized at registration (from the type's bound when the
    /// caller names none), as `SubBufferedRawInfoCEntry` has been since
    /// phase-408 W5b. It was an inline `[u8; RX_BUF]`, so every entry of this
    /// shape reserved the image's closure maximum whatever its own type was.
    pub(crate) buffer: TrailingBuf,
    pub(crate) callback: F,
}

/// Dispatch for the generic raw safety subscription: validate-receive into the
/// buffer, then pass the raw slice + status to the callback.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawSafetyEntry<F>`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_buffered_raw_safety_try_process<F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    F: FnMut(&[u8], &nros_rmw::IntegrityStatus),
{
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawSafetyEntry<F>) };
    match entry.handle.take_validated(entry.buffer.as_mut_slice()) {
        // Phase 8 — hooked. One sample per dispatch. A failed validation
        // still delivers the sample WITH its status, so this arm is the only
        // callback path; `Ok(None)` / `Err(_)` fire nothing.
        Ok(Some((len, status))) => {
            trace_take(desc_idx, &entry.buffer.as_slice()[..len], None);
            trace_cb_start(desc_idx);
            (entry.callback)(&entry.buffer.as_slice()[..len], &status);
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Readiness check for the generic raw safety subscription.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawSafetyEntry<F>`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_buffered_raw_safety_has_data<F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawSafetyEntry<F>) };
    entry.handle.has_data()
}

/// Phase 269 W3 — the C analog of [`SubBufferedRawSafetyEntry`]: same flat inline
/// payload buffer + `take_validated` dispatch, but the callback is a plain
/// C function pointer (`RawSubscriptionSafetyCallback`) that receives the integrity
/// scalars alongside the CDR bytes.
///
/// phase-408 W5b — the payload slot is a [`TrailingBuf`] sized from the
/// registering caller's `rx_buffer_hint`, so this entry is no longer generic at
/// all: the `RX_BUF` const it was monomorphised over only ever arrived as
/// `DEFAULT_RX_BUF_SIZE`, and the status a validated sample carries is
/// per-sample side data, which is why the slot stays flat rather than becoming
/// a [`BufferStrategy`].
#[cfg(feature = "safety-e2e")]
#[repr(C)]
pub(crate) struct SubBufferedRawSafetyCEntry {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: TrailingBuf,
    pub(crate) callback: super::types::RawSubscriptionSafetyCallback,
    pub(crate) context: *mut core::ffi::c_void,
}

/// Dispatch for the C-style raw validated subscription: validate-receive into the
/// buffer, then pass the raw slice + unpacked integrity scalars to the callback.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawSafetyCEntry`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_buffered_raw_safety_c_try_process(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawSafetyCEntry) };
    let payload = entry.buffer.as_mut_slice();
    match entry.handle.take_validated(payload) {
        // Phase 8 — hooked. One sample per dispatch. The `crc_valid` unpack
        // is executor bookkeeping, not user code, so it stays OUTSIDE the
        // span; the pair brackets only the FFI call.
        Ok(Some((len, status))) => {
            let crc_valid: i8 = match status.crc_valid {
                Some(true) => 1,
                Some(false) => 0,
                None => -1,
            };
            trace_take(
                desc_idx,
                unsafe { core::slice::from_raw_parts(entry.buffer.as_ptr(), len) },
                None,
            );
            trace_cb_start(desc_idx);
            unsafe {
                (entry.callback)(
                    entry.buffer.as_ptr(),
                    len,
                    status.gap,
                    status.duplicate,
                    crc_valid,
                    entry.context,
                )
            };
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Readiness check for the C-style raw validated subscription.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawSafetyCEntry`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_buffered_raw_safety_c_has_data(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawSafetyCEntry) };
    entry.handle.has_data()
}

/// Buffered subscription entry for C-style raw callbacks (function pointer + context).
///
/// Same as `SubBufferedRawEntry` but uses `RawSubscriptionCallback` instead of
/// a Rust closure. Used by the C API and by the `register_subscription_buffered_raw_*`
/// methods.
///
/// # Where the caller's capture went — phase-456 W8
///
/// Phase-456 W1 put the caller's capture INSIDE this struct, as a fixed
/// `[u8; CALLBACK_CAPTURE_BYTES]` with ~~`CALLBACK_CAPTURE_BYTES`~~ equal to a C++
/// macro's value by construction. That made a number an ABI agreement between
/// two languages, with a runtime `BufferTooSmall` as the only thing standing
/// between a disagreement and a corrupted dispatch — a failure mode reachable
/// ONLY when the two sides drifted, which is to say a failure mode nobody could
/// test against.
///
/// W8 made the capture a RUNTIME LENGTH: exactly `capture.len()` bytes of arena,
/// with `context` pointing at them. Since phase-476 W0 those bytes sit at the end
/// of the entry's own trailing region (`Executor::place_capture`), so releasing
/// the entry releases them.
/// The bound is now the arena, which every other allocation here already shares,
/// and the constant is gone from both sides of the seam. The entry also stops
/// carrying 32 bytes that every non-capturing C registration paid for.
#[repr(C)]
pub(crate) struct SubBufferedRawCEntry {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: BufferStrategy,
    pub(crate) callback: RawSubscriptionCallback,
    /// The dispatch context. Either what the caller passed, or — when a capture
    /// was supplied — a pointer into the arena's own copy of it, so nothing the
    /// caller owns is referenced after registration returns.
    pub(crate) context: *mut core::ffi::c_void,
    /// phase-476 W2 — destroys that copy when the entry is dropped. Declared
    /// after `handle`, so the subscriber is gone before its capture is.
    pub(crate) capture: CaptureGuard,
    /// Issue 2215 — the contracted age row this subscription feeds, if any.
    pub(crate) age_mon: Option<RawAgeMon>,
}

// ============================================================================
// In-place raw C subscription (phase-456 W8)
// ============================================================================

/// In-place C-callback subscription entry — **no arena buffer**.
///
/// The C-family twin of [`SubInplaceEntry`], and the one the executor should
/// have had first: `RawSubscriptionCallback` is `(const uint8_t*, size_t,
/// void*)`, which is already a BORROWED-BYTES signature, so handing it the
/// backend's own receive slot costs nothing and needs no conversion — where the
/// Rust typed path has to deserialize an owned `&M` out of the same bytes.
///
/// Until phase-456 W8 the C registration path never consulted
/// `supports_process_in_place()`, so every C and C++ subscription on zenoh or
/// XRCE allocated a receive region the backend does not need — the same order
/// as issue 1340's ~9.7 KiB per subscription, and on the language surface that
/// had the easiest time doing without it.
#[repr(C)]
pub(crate) struct SubInplaceRawCEntry {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) callback: RawSubscriptionCallback,
    /// Same contract as [`SubBufferedRawCEntry::context`].
    pub(crate) context: *mut core::ffi::c_void,
    /// Same contract as [`SubBufferedRawCEntry::capture`].
    pub(crate) capture: CaptureGuard,
    /// Same contract as [`SubBufferedRawCEntry::age_mon`].
    pub(crate) age_mon: Option<RawAgeMon>,
}

/// Dispatch for in-place C-style raw subscriptions.
///
/// Drains pending samples, up to [`IN_PLACE_DRAIN_LIMIT`], invoking the callback
/// directly on the backend's borrowed slot. Returns `Ok(true)` if any sample was dispatched.
///
/// # Safety
/// `ptr` must point to a valid, aligned [`SubInplaceRawCEntry`].
pub(crate) unsafe fn sub_inplace_raw_c_try_process(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SubInplaceRawCEntry) };
    // Split-borrow the handle from the callback + context (disjoint fields).
    let SubInplaceRawCEntry {
        handle,
        callback,
        context,
        age_mon,
        ..
    } = entry;
    let context = *context;
    let callback = *callback;
    let age_mon = *age_mon;
    drain_in_place(|| {
        // Phase 8 — hooked INSIDE the borrow closure, which the drain loop
        // re-enters once per pending sample, so N samples produce N spans. The
        // pair is fully contained in the closure, so the `?` below can never
        // fire between a start and its end.
        handle.process_raw_in_place(|raw| {
            observe_age_raw(raw, &age_mon);
            trace_take(desc_idx, raw, None);
            trace_cb_start(desc_idx);
            unsafe { callback(raw.as_ptr(), raw.len(), context) };
            trace_cb_end(desc_idx);
        })
    })
}

/// Readiness check for in-place C-style raw subscriptions.
///
/// # Safety
/// `ptr` must point to a valid [`SubInplaceRawCEntry`].
pub(crate) unsafe fn sub_inplace_raw_c_has_data(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubInplaceRawCEntry) };
    entry.handle.has_data()
}

/// In-place raw subscription entry for the Rust GENERIC path — issue 1340.
///
/// The Rust twin of [`SubInplaceRawCEntry`]: `F` is `FnMut(&[u8])`, a
/// borrowed-bytes callback, which is exactly what `process_raw_in_place` hands
/// over. So there is no receive region to reserve — the backend's own slot is
/// the buffer, for the duration of the call.
///
/// `#[repr(C)]` with `handle` first, like every other subscription entry: the
/// executor's tests reach the backend subscriber through the entry's offset,
/// and that cast is sound only while the handle leads.
///
/// Carries no age monitor, and loses none: the buffered entry it replaces on an
/// in-place backend ([`SubBufferedRawEntry`]) carries none either. The only
/// thing that does not move over is the ring, which is the point.
#[repr(C)]
pub(crate) struct SubInplaceRawEntry<F> {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) callback: F,
}

/// Dispatch for in-place Rust raw subscriptions.
///
/// Drains pending samples, up to [`IN_PLACE_DRAIN_LIMIT`], invoking the callback
/// directly on the backend's borrowed slot. Returns `Ok(true)` if any sample was dispatched.
/// Same drain and the same trace pairing as [`sub_inplace_raw_c_try_process`],
/// which it mirrors.
///
/// # Safety
/// `ptr` must point to a valid, aligned [`SubInplaceRawEntry<F>`].
pub(crate) unsafe fn sub_inplace_raw_try_process<F: FnMut(&[u8])>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SubInplaceRawEntry<F>) };
    // Split-borrow the handle from the callback (disjoint fields).
    let SubInplaceRawEntry { handle, callback } = entry;
    drain_in_place(|| {
        // Hooked INSIDE the borrow closure, which the drain loop re-enters once
        // per pending sample, so N samples produce N spans and the `?` below
        // can never fire between a start and its end.
        handle.process_raw_in_place(|raw| {
            trace_take(desc_idx, raw, None);
            trace_cb_start(desc_idx);
            callback(raw);
            trace_cb_end(desc_idx);
        })
    })
}

/// Readiness check for in-place Rust raw subscriptions.
///
/// # Safety
/// `ptr` must point to a valid [`SubInplaceRawEntry<F>`].
pub(crate) unsafe fn sub_inplace_raw_has_data<F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubInplaceRawEntry<F>) };
    entry.handle.has_data()
}

/// Drain helper for C-style raw buffered entries.
/// Issue 0737 — a transport ERROR is not "no data", and conflating them
/// destroys the sample without a trace.
///
/// Both arms used to read `if let Ok(Some(len)) = … else { break }`, which
/// treats `Err(_)` exactly like `Ok(None)`. The backend has ALREADY consumed
/// the sample by the time it reports the error, so the message is gone and the
/// only observable is that nothing arrived — indistinguishable from a publisher
/// that never published. 0737 spent two hosts' investigations inside that
/// ambiguity while the executor's own `alive — … 0 error(s)` line reported
/// health, because the error never reached the counter that prints it.
///
/// Now it propagates: `spin_once` maps an `Err` from `try_process` to
/// `subscription_errors`, so the count stops lying and the failure has a name.
unsafe fn drain_into_buffer_raw_c(entry: &mut SubBufferedRawCEntry) -> Result<(), TransportError> {
    match &entry.buffer {
        BufferStrategy::Triple(tb) => {
            let slot = tb.write_slot();
            if let Some(len) = entry.handle.take_serialized(slot)? {
                tb.writer_publish(len);
            }
        }
        BufferStrategy::Ring(ring) => {
            while let Some(slot) = ring.try_push() {
                match entry.handle.take_serialized(slot)? {
                    Some(len) => ring.commit_push(len),
                    None => break,
                }
            }
        }
    }
    Ok(())
}

/// Dispatch for C-style raw buffered subscriptions.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubBufferedRawCEntry`.
pub(crate) unsafe fn sub_buffered_raw_c_try_process(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SubBufferedRawCEntry) };

    // Issue 0737 — drain first, and let a transport error OUT. Anything already
    // buffered from an earlier spin is still dispatched below on the next call.
    unsafe { drain_into_buffer_raw_c(entry)? };

    match &entry.buffer {
        // Phase 8 — hooked. Triple-buffered: at most one invocation per
        // `try_process`, so exactly one start/end pair, and none at all when
        // `reader_acquire` comes back empty.
        BufferStrategy::Triple(tb) => match tb.reader_acquire() {
            Some((data, len)) => {
                observe_age_raw(&data[..len], &entry.age_mon);
                trace_take(desc_idx, &data[..len], None);
                trace_cb_start(desc_idx);
                unsafe { (entry.callback)(data.as_ptr(), len, entry.context) };
                trace_cb_end(desc_idx);
                Ok(true)
            }
            None => Ok(false),
        },
        // Phase 8 — hooked INSIDE the loop, deliberately. A ring drains N
        // queued messages per `try_process` and fires the user callback once
        // per message; a pair outside the loop would report N invocations as
        // one, which is the granularity failure that ruled out hooking the
        // `try_process` boundary in the first place.
        BufferStrategy::Ring(ring) => {
            let mut did_work = false;
            while let Some((data, len)) = ring.try_pop() {
                observe_age_raw(&data[..len], &entry.age_mon);
                trace_take(desc_idx, &data[..len], None);
                trace_cb_start(desc_idx);
                unsafe { (entry.callback)(data.as_ptr(), len, entry.context) };
                trace_cb_end(desc_idx);
                ring.commit_pop();
                did_work = true;
            }
            Ok(did_work)
        }
    }
}

/// Readiness check for C-style raw buffered subscriptions.
///
/// # Safety
/// `ptr` must point to a valid `SubBufferedRawCEntry`.
pub(crate) unsafe fn sub_buffered_raw_c_has_data(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedRawCEntry) };
    entry.handle.has_data() || entry.buffer.has_data()
}

// ============================================================================
// phase-417 W5.a — TYPED C subscription delivery (caller-owned storage)
// ============================================================================

/// Buffered subscription entry that deserialises into CALLER-OWNED storage
/// before dispatching (phase-417 W5.a, RFC-0089 stage 5).
///
/// Structurally [`SubBufferedRawCEntry`] plus two words: the caller's `msg`
/// pointer and the erased deserialiser for its type. Deliberately the same
/// buffering machinery — a typed subscription is a raw one with one extra step
/// between the ring and the callback, not a second arena and not a second
/// receive path.
///
/// `msg` is not owned here and is not dropped: it is the caller's storage,
/// which is exactly what makes typed delivery possible with no allocator (the
/// property rclc has and the C ledger wrongly claimed rclc lacked).
#[repr(C)]
pub(crate) struct SubBufferedTypedCEntry {
    pub(crate) handle: session::RmwSubscriber,
    pub(crate) buffer: BufferStrategy,
    /// Caller-owned message storage. Overwritten on every dispatch.
    pub(crate) msg: *mut core::ffi::c_void,
    pub(crate) deserialize: RawMessageDeserializeFn,
    pub(crate) callback: TypedSubscriptionCallback,
    pub(crate) context: *mut core::ffi::c_void,
}

/// Drain helper for typed buffered entries — the raw one, over the typed entry.
///
/// Issue 0737's rule applies unchanged: a transport ERROR is not "no data", so
/// it propagates instead of being flattened into an empty take.
unsafe fn drain_into_buffer_typed_c(
    entry: &mut SubBufferedTypedCEntry,
) -> Result<(), TransportError> {
    match &entry.buffer {
        BufferStrategy::Triple(tb) => {
            let slot = tb.write_slot();
            if let Some(len) = entry.handle.take_serialized(slot)? {
                tb.writer_publish(len);
            }
        }
        BufferStrategy::Ring(ring) => {
            while let Some(slot) = ring.try_push() {
                match entry.handle.take_serialized(slot)? {
                    Some(len) => ring.commit_push(len),
                    None => break,
                }
            }
        }
    }
    Ok(())
}

/// Deserialise one sample into the caller's storage and dispatch.
///
/// Returns `false` when the deserialiser refused the sample. The callback is
/// NOT invoked in that case — a caller that read `msg` after a failed decode
/// would be reading the previous message, or an `_init` default, under the
/// impression it was the one that just arrived. That is precisely the
/// "compile and differ" RFC-0089 forbids, so the failure costs the dispatch.
///
/// # Safety
/// `entry` must be a live typed entry; `data`/`len` a readable CDR sample.
unsafe fn typed_dispatch_one(
    entry: &SubBufferedTypedCEntry,
    data: *const u8,
    len: usize,
    desc_idx: u8,
) -> bool {
    let rc = unsafe { (entry.deserialize)(data, len, entry.msg) };
    if rc != 0 {
        report_typed_deserialize_failure(rc, len);
        return false;
    }
    trace_take(
        desc_idx,
        unsafe { core::slice::from_raw_parts(data, len) },
        None,
    );
    trace_cb_start(desc_idx);
    unsafe { (entry.callback)(entry.msg as *const core::ffi::c_void, entry.context) };
    trace_cb_end(desc_idx);
    true
}

/// Count of samples dropped because the caller's deserialiser refused them.
/// A static for the reason [`DROPPED_TAKES`] is one.
static TYPED_DESERIALIZE_FAILURES: portable_atomic::AtomicU32 = portable_atomic::AtomicU32::new(0);

/// Report a refused decode: first, then every 64th.
///
/// Rate-limited for the reason [`report_dropped_take`] is — a 40-participant
/// graph must not turn one type mismatch into a log flood (issue 0371's shape).
/// `nros_log`, never stdio (issue 0589), so this reaches a `no_std` image.
///
/// The single most likely cause is the one this message names first: the
/// generated struct's inline capacity is smaller than the sample. Both
/// `rclc_cdr_read_string` (`str_len > max_len`) and the bounded-sequence arm
/// (`len > capacity`) FAIL rather than truncate, so an oversized message
/// arrives here as a refusal and never as a short message the callback cannot
/// tell from a real one.
#[cold]
fn report_typed_deserialize_failure(rc: i32, len: usize) {
    let n = TYPED_DESERIALIZE_FAILURES.fetch_add(1, portable_atomic::Ordering::Relaxed);
    if n != 0 && !n.is_multiple_of(64) {
        return;
    }
    nros_log::log_error!(
        nros_log::get_logger("nros"),
        "typed subscription DROPPED a {len}-byte sample: deserialize returned \
         {rc}. The callback was NOT invoked and the storage still holds the \
         previous message. Usual cause: the sample does not fit the message \
         struct (a string/sequence longer than its declared bound) or the topic \
         carries a different type. Dropped {} so far (phase-417 W5.a)",
        n + 1
    );
}

/// Test-only accessor for [`TYPED_DESERIALIZE_FAILURES`].
///
/// The cfg MATCHES `mod tests`' own (`executor/mod.rs:95`), not a bare
/// `#[cfg(test)]`: the test module is `not(feature = "rmw-cffi")` too, so under
/// `test + rmw-cffi` a bare gate leaves this compiled with its only caller
/// gone, and `-D warnings` turns that into a build failure rather than a lint.
/// An accessor's gate has to be the gate of the thing that reads it — and
/// that gate gained `alloc` after this one was written, so under default
/// features this was compiled with no caller at all.
#[cfg(all(test, feature = "alloc", not(feature = "rmw-cffi")))]
pub(crate) fn typed_deserialize_failures() -> u32 {
    TYPED_DESERIALIZE_FAILURES.load(portable_atomic::Ordering::Relaxed)
}

/// Dispatch for typed C subscriptions.
///
/// # Safety
/// `ptr` must point to a valid, aligned [`SubBufferedTypedCEntry`].
pub(crate) unsafe fn sub_buffered_typed_c_try_process(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SubBufferedTypedCEntry) };

    unsafe { drain_into_buffer_typed_c(entry)? };

    match &entry.buffer {
        BufferStrategy::Triple(tb) => match tb.reader_acquire() {
            Some((data, len)) => {
                if unsafe { typed_dispatch_one(entry, data.as_ptr(), len, desc_idx) } {
                    Ok(true)
                } else {
                    Err(TransportError::DeserializationError)
                }
            }
            None => Ok(false),
        },
        // A refused sample ENDS the drain and is reported, rather than being
        // skipped so the loop can keep going: `try_process` carries one verdict
        // per call, and a `Ok(true)` that also swallowed a decode failure would
        // never reach `subscription_errors`. The samples still queued behind it
        // are dispatched on the next spin — the ring is not cleared.
        BufferStrategy::Ring(ring) => {
            let mut did_work = false;
            while let Some((data, len)) = ring.try_pop() {
                let ok = unsafe { typed_dispatch_one(entry, data.as_ptr(), len, desc_idx) };
                ring.commit_pop();
                if !ok {
                    return Err(TransportError::DeserializationError);
                }
                did_work = true;
            }
            Ok(did_work)
        }
    }
}

/// Readiness check for typed C subscriptions.
///
/// # Safety
/// `ptr` must point to a valid [`SubBufferedTypedCEntry`].
pub(crate) unsafe fn sub_buffered_typed_c_has_data(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubBufferedTypedCEntry) };
    entry.handle.has_data() || entry.buffer.has_data()
}

// ============================================================================
// Dispatch functions
// ============================================================================

/// Monomorphized subscription dispatch function (with MessageInfo).
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubInfoEntry<M, F>`.
pub(crate) unsafe fn sub_info_try_process<M, F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    M: RosMessage,
    F: FnMut(&M, Option<&MessageInfo>),
{
    let entry = unsafe { &mut *(ptr as *mut SubInfoEntry<M, F>) };

    // LET mode: use pre-sampled data if available (no MessageInfo in snapshot)
    //
    // Phase 8 — hooked. This is a SECOND callback site, mutually exclusive
    // with the receive path below: LET mode dispatches the snapshot and
    // returns, so a dispatch emits one pair from here OR one from there,
    // never both.
    if entry.sampled_len > 0 {
        let len = entry.sampled_len;
        entry.sampled_len = 0;
        let mut reader = CdrReader::new_with_header(&entry.buffer.as_slice()[..len])
            .map_err(|_| TransportError::DeserializationError)?;
        let msg = M::deserialize(&mut reader).map_err(|_| TransportError::DeserializationError)?;
        trace_take(
            desc_idx,
            &entry.buffer.as_slice()[..len],
            <M as RosMessage>::STAMP_OFFSET,
        );
        trace_cb_start(desc_idx);
        (entry.callback)(&msg, None);
        trace_cb_end(desc_idx);
        return Ok(true);
    }

    match entry
        .handle
        .take_serialized_with_info(entry.buffer.as_mut_slice())
    {
        // Phase 8 — hooked. One sample per dispatch; `Ok(None)` / `Err(_)`
        // fire nothing, and both `?`s precede the start hook.
        Ok(Some((len, info))) => {
            let mut reader = CdrReader::new_with_header(&entry.buffer.as_slice()[..len])
                .map_err(|_| TransportError::DeserializationError)?;
            let msg =
                M::deserialize(&mut reader).map_err(|_| TransportError::DeserializationError)?;
            trace_take(
                desc_idx,
                &entry.buffer.as_slice()[..len],
                <M as RosMessage>::STAMP_OFFSET,
            );
            trace_cb_start(desc_idx);
            (entry.callback)(&msg, info.as_ref());
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Monomorphized subscription dispatch function (with safety validation).
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubSafetyEntry<M, F>`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_safety_try_process<M, F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    M: RosMessage,
    F: FnMut(&M, &nros_rmw::IntegrityStatus),
{
    let entry = unsafe { &mut *(ptr as *mut SubSafetyEntry<M, F>) };

    // LET mode: use pre-sampled data (no IntegrityStatus in snapshot)
    //
    // Phase 8 — hooked. A SECOND callback site, mutually exclusive with the
    // validated-receive path below (this one `return`s). The synthetic
    // all-clear `IntegrityStatus` is built inside the span because it is an
    // argument expression, not user code — it is three constant stores.
    if entry.sampled_len > 0 {
        let len = entry.sampled_len;
        entry.sampled_len = 0;
        let mut reader = CdrReader::new_with_header(&entry.buffer.as_slice()[..len])
            .map_err(|_| TransportError::DeserializationError)?;
        let msg = M::deserialize(&mut reader).map_err(|_| TransportError::DeserializationError)?;
        trace_take(
            desc_idx,
            &entry.buffer.as_slice()[..len],
            <M as RosMessage>::STAMP_OFFSET,
        );
        trace_cb_start(desc_idx);
        (entry.callback)(
            &msg,
            &nros_rmw::IntegrityStatus {
                gap: 0,
                duplicate: false,
                crc_valid: None,
            },
        );
        trace_cb_end(desc_idx);
        return Ok(true);
    }

    match entry.handle.take_validated(entry.buffer.as_mut_slice()) {
        // Phase 8 — hooked. One sample per dispatch; both `?`s precede the
        // start hook, and `Ok(None)` / `Err(_)` fire nothing.
        Ok(Some((len, status))) => {
            let mut reader = CdrReader::new_with_header(&entry.buffer.as_slice()[..len])
                .map_err(|_| TransportError::DeserializationError)?;
            let msg =
                M::deserialize(&mut reader).map_err(|_| TransportError::DeserializationError)?;
            trace_take(
                desc_idx,
                &entry.buffer.as_slice()[..len],
                <M as RosMessage>::STAMP_OFFSET,
            );
            trace_cb_start(desc_idx);
            (entry.callback)(&msg, &status);
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => Err(TransportError::DeserializationError),
    }
}

/// Monomorphized service dispatch function.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SrvEntry<Svc, F, REQ_BUF, REPLY_BUF>`.
pub(crate) unsafe fn srv_try_process<Svc, F, const REQ_BUF: usize, const REPLY_BUF: usize>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    Svc: RosService,
    F: FnMut(&Svc::Request) -> Svc::Reply,
{
    let entry = unsafe { &mut *(ptr as *mut SrvEntry<Svc, F, REQ_BUF, REPLY_BUF>) };
    // Split borrow: destructure entry to avoid aliasing issues
    let SrvEntry {
        handle,
        req_buffer,
        reply_buffer,
        callback,
        ..
    } = entry;
    handle
        // Phase 8 — hooked INSIDE the closure `handle_request` invokes, not
        // around `handle_request` itself. The closure runs once per request
        // actually received, so a spin with nothing pending emits nothing;
        // bracketing the outer call would have timed the receive/reply
        // machinery and fired on every empty poll.
        .handle_request::<Svc>(req_buffer, reply_buffer, |req| {
            trace_cb_start(desc_idx);
            let reply = (callback)(req);
            trace_cb_end(desc_idx);
            reply
        })
        .map_err(|_| TransportError::ServiceReplyFailed)
}

/// Issue 1942 — [`srv_try_process`] for a callback that also takes the
/// request's [`ServiceInfo`](nros_core::ServiceInfo) (rclrs's shape). Same
/// [`SrvEntry`] layout; only the callback's signature differs.
///
/// # Safety
/// `ptr` must point to a valid `SrvEntry<Svc, F, REQ_BUF, REPLY_BUF>`.
pub(crate) unsafe fn srv_try_process_with_info<
    Svc,
    F,
    const REQ_BUF: usize,
    const REPLY_BUF: usize,
>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    Svc: RosService,
    F: FnMut(&Svc::Request, &nros_core::ServiceInfo) -> Svc::Reply,
{
    let entry = unsafe { &mut *(ptr as *mut SrvEntry<Svc, F, REQ_BUF, REPLY_BUF>) };
    let SrvEntry {
        handle,
        req_buffer,
        reply_buffer,
        callback,
        ..
    } = entry;
    handle
        // Traced inside the closure, as in `srv_try_process`.
        .handle_request_with_info::<Svc>(req_buffer, reply_buffer, |req, info| {
            trace_cb_start(desc_idx);
            let reply = (callback)(req, info);
            trace_cb_end(desc_idx);
            reply
        })
        .map_err(|_| TransportError::ServiceReplyFailed)
}

/// Monomorphized drop function for arena entries.
///
/// # Safety
/// `ptr` must point to a valid, aligned `T` that has not been dropped.
pub(crate) unsafe fn drop_entry<T>(ptr: *mut u8) {
    unsafe { core::ptr::drop_in_place(ptr as *mut T) };
}

/// What one executor cycle does to a timer's own `elapsed_us`.
pub(crate) enum TimerStep {
    /// Add this many microseconds, measured on the TIMER'S clock.
    Advance(u64),
    /// The timer's clock jumped BACKWARDS — a bag looping, a simulator reset,
    /// an NTP step. Restart the period rather than stalling the timer for the
    /// length of the jump, which is what accumulating a negative delta would
    /// amount to. rclcpp gets here through a jump callback; we need the
    /// behaviour, not (yet) the callback surface
    /// (`c:clock_add_jump_callback` stays declined).
    Restart,
}

/// One cycle's advance for a timer, in the timer's OWN clock — phase-425 W4,
/// made shared by issue 1321.
///
/// `Steady` consumes the executor's spin delta, unchanged and free. The other
/// two READ their clock and diff against the last reading, which is what makes
/// a paused simulator pause the timer: `/clock` stops advancing, so the delta
/// is zero. A FORWARD jump is deliberately NOT special-cased: it lands in
/// `elapsed_us` as a backlog, and the overrun policy is the documented
/// mechanism for deciding whether a backlog replays (`CatchUp`) or coalesces
/// (`Skip`).
///
/// One spelling, because there are two callers: the dispatcher below, and the
/// budget-skip path in `spin_once`. That second one had the delta computation
/// open-coded as "add the executor's `delta_us`", which is a WALL quantity —
/// the same units error as issue 1321's park bound, but unbounded: it advanced
/// a `Ros` timer while `/clock` was paused, which is the one thing a ROS-time
/// timer must never do.
pub(crate) fn timer_clock_step(
    clock_source: TimerClockSource,
    last_clock_ns: &mut i64,
    spin_delta_us: u64,
) -> TimerStep {
    match clock_source {
        TimerClockSource::Steady => TimerStep::Advance(spin_delta_us),
        source => {
            let now_ns = source.now_ns();
            let step_ns = now_ns - *last_clock_ns;
            *last_clock_ns = now_ns;
            if step_ns < 0 {
                return TimerStep::Restart;
            }
            TimerStep::Advance((step_ns as u64) / 1_000)
        }
    }
}

/// Advance a timer's own sense of time by one cycle WITHOUT dispatching it —
/// issue 0736's accounting, on issue 1321's clock.
///
/// Type-erased because its caller has a `TimerHeader`, not a `TimerEntry<F>`.
pub(crate) fn timer_advance_without_dispatch(header: &mut TimerHeader, spin_delta_us: u64) {
    match timer_clock_step(
        header.clock_source,
        &mut header.last_clock_ns,
        spin_delta_us,
    ) {
        TimerStep::Advance(us) => header.elapsed_us = header.elapsed_us.saturating_add(us),
        TimerStep::Restart => header.elapsed_us = 0,
    }
}

/// A `Skip` timer's overrun after it fired: the periods it missed beyond the
/// one that fired, and what stays of `elapsed_us` on the phase grid.
///
/// Saturating, because `elapsed_us` can be BELOW one period here: a callback
/// that reset its own timer (`timer->reset()`, which upstream allows inside
/// the callback) zeroed it while the timer was firing. `0 / period - 1`
/// panicked a debug build (phase-518 W2: issue 2055's callback pump re-arms
/// its timer from a callback that sends again).
pub(crate) fn skip_overrun(elapsed_us: u64, period_us: u64) -> (u64, u64) {
    (
        (elapsed_us / period_us).saturating_sub(1),
        elapsed_us % period_us,
    )
}

/// The call info for an activation that is due NOW — issue 1974.
///
/// `actual` is now on the timer's clock ([`TimerClockSource::dispatch_now_ns`]);
/// `expected` subtracts how far `elapsed_us` has run past `period_us`. A
/// zero-period timer ("fire every spin") has no deadline to be late for, so its
/// two times coincide.
pub(crate) fn timer_call_info(
    clock_source: TimerClockSource,
    last_clock_ns: i64,
    elapsed_us: u64,
    period_us: u64,
) -> TimerCallInfo {
    let actual = clock_source.dispatch_now_ns(last_clock_ns);
    let late_us = if period_us == 0 {
        0
    } else {
        elapsed_us.saturating_sub(period_us)
    };
    let late_ns = i64::try_from(late_us.saturating_mul(1_000)).unwrap_or(i64::MAX);
    TimerCallInfo {
        expected_call_time: actual.saturating_sub(late_ns),
        actual_call_time: actual,
        clock: clock_source,
        since_last_call_ns: i64::try_from(elapsed_us.saturating_mul(1_000)).unwrap_or(i64::MAX),
    }
}

/// Monomorphized timer dispatch function.
///
/// # Safety
/// `ptr` must point to a valid, aligned `TimerEntry<F>`.
pub(crate) unsafe fn timer_try_process<F>(
    ptr: *mut u8,
    delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    F: FnMut(&TimerCallInfo),
{
    let entry = unsafe { &mut *(ptr as *mut TimerEntry<F>) };

    // Cancelled or one-shot already fired
    if entry.cancelled || (entry.oneshot && entry.fired) {
        return Ok(false);
    }

    // phase-425 W4 — which clock advanced, and by how much. Shared with the
    // budget-skip path in `spin_once`, which advances the same counter without
    // dispatching (issue 0736): there is ONE answer to "which clock moves this
    // timer", and it is `timer_clock_step`.
    let delta_us = match timer_clock_step(entry.clock_source, &mut entry.last_clock_ns, delta_us) {
        TimerStep::Advance(us) => us,
        TimerStep::Restart => {
            entry.elapsed_us = 0;
            return Ok(false);
        }
    };

    entry.elapsed_us = entry.elapsed_us.saturating_add(delta_us);

    if entry.elapsed_us >= entry.period_us {
        // Phase 8 — hooked. Inside the due-check, so a timer polled on every
        // spin and not yet due emits nothing: `Ok(false)` is the COMMON
        // outcome here, and it is exactly the over-reporting that a hook at
        // the `try_process` boundary would have produced.
        // phase-474 I3 -- one tick in N when the timer is thinned.
        let traced = timer_tick_traced(desc_idx);
        // Issue 1974 — the ONE producer of a timer's call info, computed from
        // the very comparison that just decided the activation is due, so the
        // three language surfaces read the dispatcher's answer rather than
        // keeping their own. Lateness is how far `elapsed_us` ran past the
        // period; it is never negative here, so `actual >= expected` always.
        let info = timer_call_info(
            entry.clock_source,
            entry.last_clock_ns,
            entry.elapsed_us,
            entry.period_us,
        );
        if traced {
            trace_cb_start(desc_idx);
        }
        (entry.callback)(&info);
        if traced {
            trace_cb_end(desc_idx);
        }
        if entry.oneshot {
            entry.fired = true;
        } else if entry.period_us == 0 {
            entry.elapsed_us = 0;
        } else {
            // Issue #505 — a stall leaves `elapsed_us` worth several
            // periods. `CatchUp` subtracts one period per pass, so the
            // backlog replays back-to-back; `Skip` drops the whole
            // backlog in one step and counts it, keeping the remainder
            // so activations stay on the original phase grid.
            match entry.overrun_policy {
                TimerOverrunPolicy::CatchUp => {
                    entry.elapsed_us = entry.elapsed_us.saturating_sub(entry.period_us);
                }
                TimerOverrunPolicy::Skip => {
                    let (missed, remainder) = skip_overrun(entry.elapsed_us, entry.period_us);
                    if missed > 0 {
                        entry.overruns = entry
                            .overruns
                            .saturating_add(u32::try_from(missed).unwrap_or(u32::MAX));
                    }
                    entry.elapsed_us = remainder;
                }
            }
        }
        Ok(true)
    } else {
        Ok(false)
    }
}

/// Monomorphized action server dispatch function.
///
/// Polls goal acceptance, cancel handling, and result serving.
///
/// # Safety
/// `ptr` must point to a valid, aligned `ActionServerArenaEntry<...>`.
pub(crate) unsafe fn action_server_try_process<
    A,
    GoalF,
    CancelF,
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
    const MAX_GOALS: usize,
>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    A: RosAction,
    A::Goal: Clone,
    A::Result: Clone + Default,
    GoalF: FnMut(&nros_core::GoalUuid, &A::Goal) -> nros_core::GoalResponse,
    CancelF: FnMut(&nros_core::GoalUuid, nros_core::GoalStatusCode) -> nros_core::CancelResponse,
{
    let entry = unsafe {
        &mut *(ptr as *mut ActionServerArenaEntry<
            A,
            GoalF,
            CancelF,
            GOAL_BUF,
            RESULT_BUF,
            FEEDBACK_BUF,
            MAX_GOALS,
        >)
    };
    let ActionServerArenaEntry {
        server,
        goal_callback,
        cancel_callback,
    } = entry;

    let mut did_work = false;

    // Handle cancels first
    //
    // Phase 8 — hooked INSIDE the closure, not around `try_handle_cancel`.
    // The core runs this closure only when a cancel request is actually
    // pending; `Ok(None)` with no user code run is the COMMON outcome, so a
    // pair at the call boundary would report an invocation on every spin.
    if matches!(
        server.try_handle_cancel(|id, st| {
            trace_cb_start(desc_idx);
            let resp = (cancel_callback)(id, st);
            trace_cb_end(desc_idx);
            resp
        }),
        Ok(Some(_))
    ) {
        did_work = true;
    }

    // Handle new goals
    //
    // Phase 8 — hooked inside the closure for the same reason. This is a
    // DISTINCT user callback from the cancel one above and gets its own
    // pair, so the two are counted separately rather than merged.
    if matches!(
        server.try_accept_goal(|id, g| {
            trace_cb_start(desc_idx);
            let resp = (goal_callback)(id, g);
            trace_cb_end(desc_idx);
            resp
        }),
        Ok(Some(_))
    ) {
        did_work = true;
    }

    // Handle result requests
    if matches!(server.try_handle_get_result(), Ok(Some(_))) {
        did_work = true;
    }

    Ok(did_work)
}

/// Monomorphized raw action server dispatch function.
///
/// Polls goal acceptance, cancel handling, and result serving using raw bytes.
///
/// # Safety
/// `ptr` must point to a valid, aligned `ActionServerRawArenaEntry<...>`.
pub(crate) unsafe fn action_server_raw_try_process<
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
    const MAX_GOALS: usize,
>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe {
        &mut *(ptr as *mut ActionServerRawArenaEntry<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>)
    };
    let ActionServerRawArenaEntry {
        core,
        goal_callback,
        cancel_callback,
        accepted_callback,
        context,
    } = entry;

    let mut did_work = false;

    // Handle cancels first
    //
    // Phase 8 — hooked INSIDE the closure: the core runs it only when a cancel
    // request is actually pending, so nothing is emitted on the common
    // `Ok(None)` spin.
    if let Ok(Some(_)) = core.try_handle_cancel(|id, st| {
        trace_cb_start(desc_idx);
        let resp = unsafe { (*cancel_callback)(id, st, *context) };
        trace_cb_end(desc_idx);
        resp
    }) {
        did_work = true;
    }

    // Handle new goals
    if let Ok(Some(raw_req)) = core.try_recv_goal_request() {
        let goal_data = &core.goal_buffer()[..raw_req.data_len];
        // Phase 8 — hooked inside the `Ok(Some(..))` arm: no goal request,
        // no goal callback, nothing emitted.
        trace_cb_start(desc_idx);
        let response = unsafe {
            (*goal_callback)(
                &raw_req.goal_id,
                goal_data.as_ptr(),
                raw_req.data_len,
                *context,
            )
        };
        trace_cb_end(desc_idx);

        if response.is_accepted() {
            // Send the accept reply *before* running any long-running
            // post-accept hook so the client observes acceptance promptly.
            let _ = core.accept_goal(raw_req.goal_id, raw_req.sequence_number);
            // Phase 8 — the post-accept hook is a THIRD, separate user
            // callback: optional, and run only on the accepted branch, so
            // it carries its own pair rather than being folded into the
            // goal callback's span above.
            if let Some(post) = *accepted_callback {
                trace_cb_start(desc_idx);
                unsafe { post(&raw_req.goal_id, *context) };
                trace_cb_end(desc_idx);
            }
        } else {
            let _ = core.reject_goal(raw_req.sequence_number);
        }
        did_work = true;
    }

    // Handle result requests (empty default result for raw API)
    if let Ok(Some(_)) = core.try_handle_get_result_raw(&[]) {
        did_work = true;
    }

    Ok(did_work)
}

/// Monomorphized raw action client dispatch function.
///
/// Polls the action client core's non-blocking methods:
/// 1. Goal acceptance reply (`try_recv_send_goal_reply`)
/// 2. Feedback (`try_recv_feedback_raw`)
/// 3. Result reply (`try_recv_get_result_reply`)
///
/// Invokes the corresponding callback when data is available.
///
/// # Safety
/// `ptr` must point to a valid, aligned `ActionClientRawArenaEntry<...>`.
/// RFC-0069 / issues 0418 + 0035 — **retired as a correctness mechanism.**
///
/// This sniffed whether a payload already began with a CDR encapsulation header
/// (`00 <id> <opts> <opts>`) and read it directly if so. That is a VALUE test,
/// not a framing test, and it cannot tell the two apart: a leading `uint32` of
/// 256 serializes little-endian as `00 01 00 00` — byte for byte the LE encap
/// header. A sequence of length 256 is enough.
///
/// It was harmless while only Cyclone took the false branch. 0418 stopped the
/// producer writing an inner header, so EVERY payload began consulting it, and
/// a body whose first word happened to look like a header had that word eaten
/// as framing — issue #35's "sequence deserialized to len 0", reached through a
/// payload value instead of a framing bug.
///
/// Kept only for the pre-0418 diagnostic below, never for a decode decision.
#[cfg(test)]
fn payload_has_cdr_encap(p: &[u8]) -> bool {
    p.len() >= 4 && p[0] == 0x00 && matches!(p[1], 0x00 | 0x01 | 0x06 | 0x07 | 0x0a | 0x0b)
}

/// Deserialize an action result/feedback field payload, restoring the per-message
/// CDR encapsulation header the backend's typed framing may have stripped (#175).
/// `raw` is the field bytes at the payload offset; `top_encap` is the enclosing
/// message's leading 4-byte encap (always a valid header). When `raw` already
/// begins with an encap (zenoh/XRCE) it is read directly; when it does not
/// (Cyclone `dds_stream` drops the inner encap of a nested message field) the
/// top-level encap is spliced in front into a `CAP`-byte scratch buffer first.
fn read_action_field<M: nros_serdes::Deserialize, const CAP: usize>(
    top_encap: &[u8],
    raw: &[u8],
) -> Option<M> {
    // ALWAYS splice (RFC-0069 / issue 0418). Post-0418 the producer writes no
    // inner encapsulation header, and neither does Cyclone's `dds_stream` (it
    // drops the inner encap of a nested field, #175) — so the payload is field
    // bytes in every case and the enclosing message's encap is the right one to
    // read them with.
    //
    // This used to branch on `payload_has_cdr_encap(raw)`. That sniff is a value
    // test and gets it wrong for a body starting `00 01 00 00` (a `uint32` of
    // 256, e.g. a 256-element sequence), eating the leading data word — see the
    // retired helper above.
    if top_encap.len() < 4 || raw.len() + 4 > CAP {
        return None;
    }
    let mut buf = [0u8; CAP];
    buf[0..4].copy_from_slice(&top_encap[0..4]);
    buf[4..4 + raw.len()].copy_from_slice(raw);
    let mut reader = CdrReader::new_with_header(&buf[..4 + raw.len()]).ok()?;
    M::deserialize(&mut reader).ok()
}

pub(crate) unsafe fn action_client_raw_try_process<
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe {
        &mut *(ptr as *mut ActionClientRawArenaEntry<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>)
    };
    let ActionClientRawArenaEntry {
        core,
        goal_response_callback,
        feedback_callback,
        result_callback,
        cancel_response_callback,
        context,
    } = entry;

    let mut did_work = false;

    // 1. Poll goal acceptance reply
    // Issue 2009 — the goal is the one the reply's request was ABOUT. It used
    // to be rebuilt from `goal_counter`, i.e. the most recently SENT goal, so
    // two goals in flight swapped acceptances.
    if let Ok(Some((total_len, goal_id))) = core.try_recv_send_goal_reply_for() {
        if let Some(cb) = goal_response_callback {
            // Reply CDR: header (4) + accepted (u8) + stamp
            let accepted = crate::action_frame::goal_accepted(&core.result_buffer[..total_len]);
            // Phase 8 — hooked at the invocation, inside both the
            // `Ok(Some(..))` arm and the `Some(cb)` guard: an unregistered
            // callback or an empty poll emits nothing.
            trace_cb_start(desc_idx);
            unsafe { cb(&goal_id, accepted, *context) };
            trace_cb_end(desc_idx);
        }
        did_work = true;
    }

    // 2. Poll feedback
    if let Ok(Some((goal_id, total_len))) = core.try_recv_feedback_raw() {
        if let Some(cb) = feedback_callback {
            // Layout `[encap 4][GoalUuid.uuid 16][fields]` — see `action_frame`.
            // Issue 2014 — the offset (`action_frame::FEEDBACK_PAYLOAD_OFFSET`)
            // and the splice are spelled once, in `action_frame`.
            //
            // #175 + RFC-0069/0418 — the payload is field bytes on every
            // backend now: Cyclone's `dds_stream` drops the inner encap of a
            // nested field, and since 0418 the producer never writes one. So
            // splice the enclosing message's encap unconditionally. This used
            // to branch on an encap SNIFF, which mis-fires on a body whose
            // first word is 0x00010000 (see the retired helper).
            let mut spliced = [0u8; FEEDBACK_BUF];
            if let Some(n) = crate::action_frame::splice_feedback(
                &core.feedback_buffer[..total_len],
                &mut spliced,
            ) {
                // Phase 8 — a DISTINCT callback from the goal-response one,
                // with its own pair. Hooked inside the guard: a short
                // payload runs no user code and must emit nothing.
                trace_cb_start(desc_idx);
                unsafe { cb(&goal_id, spliced.as_ptr(), n, *context) };
                trace_cb_end(desc_idx);
            }
        }
        did_work = true;
    }

    // 3. Poll result reply — issue 2009: attributed to the goal its request
    //    named, not to the most recently sent goal.
    if let Ok(Some((total_len, goal_id))) = core.try_recv_get_result_reply_for() {
        if let Some(cb) = result_callback {
            // Layout `[encap 4][status i8][pad → 8][fields]` — see
            // `action_frame`. Slicing at 5 (Phase 96.1) leaked the three pad
            // bytes into the payload and emptied every C++/XRCE result.
            // Issue 2014 — status, payload offset and splice come from
            // `action_frame`, the one spelling every consumer shares.
            //
            // #175 + RFC-0069/0418 — restore the result's CDR encapsulation
            // header. The fields arrive raw on every backend now: Cyclone sends
            // `GetResult_Response` as a TYPED sample whose `result` is a NESTED
            // field, so `dds_stream` consumes the inner encap; and since 0418
            // the producer never writes one. The consumer
            // (`CallbackCtx::message` / `ffi_deserialize`) reads with
            // `new_with_header`, so splice the reply's top-level encap in front
            // unconditionally. It used to branch on an encap SNIFF, which a
            // leading `int32` of 256 (`00 01 00 00`) defeats — see the retired
            // `payload_has_cdr_encap`.
            let reply = &core.result_buffer[..total_len];
            let mut spliced = [0u8; RESULT_BUF];
            if let (Some(status), Some(n)) = (
                crate::action_frame::result_status(reply),
                crate::action_frame::splice_result(reply, &mut spliced),
            ) {
                // Phase 8 — the third distinct callback on this entry, its
                // own pair. Inside the guard, which can skip the call entirely.
                trace_cb_start(desc_idx);
                unsafe { cb(&goal_id, status, spliced.as_ptr(), n, *context) };
                trace_cb_end(desc_idx);
            }
        }
        did_work = true;
    }

    // 4. Poll cancel reply — issue 2030. Only when a callback is registered:
    //    a surface with no callback must find the reply still there. Taking it
    //    retires its request from `pending_cancel`, which is what frees the
    //    slot; before this, nothing on the C callback tier took one and the
    //    fifth `rclc_action_cancel_goal` was refused. Issue 2034 — the request's
    //    sequence id and the whole frame go along, so C++ can hand the reply to
    //    the future waiting on THAT request, with its `goals_canceling` list.
    if let Some(cb) = cancel_response_callback
        && let Ok(Some((total_len, seq, goal_id))) = core.try_recv_cancel_reply_seq()
    {
        let reply = &core.result_buffer[..total_len];
        let code = crate::action_frame::cancel_return_code(reply)
            .and_then(nros_core::CancelResponseCode::from_i8)
            .unwrap_or(nros_core::CancelResponseCode::Rejected);
        let n = crate::action_frame::cancel_goals_count(reply).unwrap_or(0);
        // Phase 8 — the fourth distinct callback on this entry, its own pair.
        trace_cb_start(desc_idx);
        unsafe {
            cb(
                &goal_id,
                seq,
                code,
                n,
                reply.as_ptr(),
                reply.len(),
                *context,
            )
        };
        trace_cb_end(desc_idx);
        did_work = true;
    }

    Ok(did_work)
}

/// Monomorphized raw service-client dispatch function.
///
/// Checks `reply_ready` (set by the transport waker) before calling
/// `take_response_raw`. This avoids blind polling on every spin tick —
/// the only cost per tick is an atomic load when no reply is pending.
///
/// # Safety
/// `ptr` must point to a valid, aligned `ServiceClientRawArenaEntry<REPLY_BUF>`.
pub(crate) unsafe fn service_client_raw_try_process<const REPLY_BUF: usize>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    use core::sync::atomic::Ordering;
    let entry = unsafe { &mut *(ptr as *mut ServiceClientRawArenaEntry<REPLY_BUF>) };

    if !entry.pending {
        return Ok(false);
    }
    // phase-492 W4 — a client registered WITHOUT a response callback is the
    // polling shape (`rclc_client_take_response`): its reply belongs to that
    // call. Taking it here dropped it, because there was no callback to hand
    // it to and `pending` was cleared — so a C component that sent a request
    // and polled from a timer never saw a single reply while the server
    // answered every one. Leave it in the handle for the poll.
    if entry.callback.is_none() && entry.request_id_callback.is_none() {
        return Ok(false);
    }

    // Clear the waker flag if set (consumed by this check).
    entry.reply_ready.store(false, Ordering::Release);

    // Issue 2009 — dispatched only if the reply names the call in flight; a
    // late reply to an earlier call is dropped here, not handed to this one.
    let ServiceClientRawArenaEntry {
        handle,
        reply_buffer,
        sequences,
        ..
    } = &mut *entry;
    match take_client_reply(handle, reply_buffer, sequences) {
        Ok(Some((len, seq))) => {
            entry.pending = false;
            // Issue 2136 — the request-id shape, when registered, in place of
            // the plain one. The id is read before anything else takes on
            // this handle, so it describes this reply.
            if let Some(cb) = entry.request_id_callback {
                let mut id = RawRequestId::for_response(
                    &nros_rmw::ClientTrait::response_info(&entry.handle),
                    seq,
                );
                trace_cb_start(desc_idx);
                unsafe { cb(entry.reply_buffer.as_ptr(), len, &mut id, entry.context) };
                trace_cb_end(desc_idx);
                return Ok(true);
            }
            // Phase 8 — hooked inside the `Some(cb)` arm. A client with no
            // callback never reaches here (see above); the not-pending and
            // `Ok(None)` paths return before this point too.
            if let Some(cb) = entry.callback {
                trace_cb_start(desc_idx);
                unsafe { cb(entry.reply_buffer.as_ptr(), len, entry.context) };
                trace_cb_end(desc_idx);
            }
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => {
            entry.pending = false;
            Err(TransportError::ServiceRequestFailed)
        }
    }
}

/// RFC-0041 / Phase 239.1 — F-independent prefix of a typed service-client
/// callback entry. `#[repr(C)]` guarantees it is the leading member of every
/// [`ServiceClientCallbackEntry`] regardless of the closure type `F`, so a
/// `ServiceClientCallback` handle can hold a `*mut` to it and send requests
/// (serialize → `send_request_raw` → set `pending`) without naming `F`.
#[repr(C)]
pub struct ServiceClientSendHeader<const REPLY_BUF: usize> {
    pub handle: session::RmwServiceClient,
    pub reply_buffer: [u8; REPLY_BUF],
    pub pending: bool,
    /// Set by the transport waker when a reply arrives (mirrors the raw entry).
    pub reply_ready: core::sync::atomic::AtomicBool,
    /// Issue 2009 — the id of the call in flight (mirrors the raw entry).
    pub sequences: ClientSequences,
}

/// Typed service-client callback entry (RFC-0041, Phase 239.1). The executor
/// eager-drains the reply at `spin_once` and dispatches it as a deserialized
/// `Svc::Reply` to the user closure — the typed analogue of
/// [`ServiceClientRawArenaEntry`]. The send side goes through the embedded
/// [`ServiceClientSendHeader`] (offset 0) via a `ServiceClientCallback` handle.
#[repr(C)]
pub(crate) struct ServiceClientCallbackEntry<Svc: RosService, F, const REPLY_BUF: usize> {
    pub(crate) hdr: ServiceClientSendHeader<REPLY_BUF>,
    pub(crate) callback: F,
    pub(crate) _phantom: PhantomData<Svc>,
}

/// Issue 2148 — what a typed callback client hands each reply to. Two shapes:
/// [`ReplyOnly`] (`FnMut(&Reply)`, `create_client_with_callback`) and
/// [`ReplyWithInfo`] (`FnMut(&Reply, &ServiceInfo)`,
/// `create_client_with_callback_with_info`). One dispatch serves both; the
/// info is read only for the shape that asks for it.
pub(crate) trait ClientReplyCallback<R> {
    /// Run the user closure. `info` reads the reply's
    /// [`ServiceInfo`](nros_core::ServiceInfo) from the handle that just took
    /// it; it is called only by a shape that wants the info.
    fn on_reply(&mut self, reply: &R, info: impl FnOnce() -> nros_core::ServiceInfo);
}

/// Issue 2148 — the reply-only shape (`FnMut(&Reply)`).
pub(crate) struct ReplyOnly<F>(pub(crate) F);

impl<R, F: FnMut(&R)> ClientReplyCallback<R> for ReplyOnly<F> {
    fn on_reply(&mut self, reply: &R, _info: impl FnOnce() -> nros_core::ServiceInfo) {
        (self.0)(reply)
    }
}

/// Issue 2148 — the reply-plus-info shape (`FnMut(&Reply, &ServiceInfo)`),
/// rclrs's `(Response, ServiceInfo)` and upstream `rmw_take_response`'s
/// `rmw_service_info_t`, as [`Promise::take_with_info`](super::handles::Promise::take_with_info)
/// reports it.
pub(crate) struct ReplyWithInfo<F>(pub(crate) F);

impl<R, F: FnMut(&R, &nros_core::ServiceInfo)> ClientReplyCallback<R> for ReplyWithInfo<F> {
    fn on_reply(&mut self, reply: &R, info: impl FnOnce() -> nros_core::ServiceInfo) {
        let info = info();
        (self.0)(reply, &info)
    }
}

/// Monomorphized typed service-client dispatch (RFC-0041, Phase 239.1).
///
/// Mirrors [`service_client_raw_try_process`] but deserializes the reply into
/// `Svc::Reply` and invokes the typed closure. Single in-flight request gated by
/// `hdr.pending`; the reply view is dropped before return (no escape).
///
/// # Safety
/// `ptr` must point to a valid, aligned `ServiceClientCallbackEntry<Svc, F, REPLY_BUF>`.
pub(crate) unsafe fn service_client_callback_try_process<Svc, F, const REPLY_BUF: usize>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    Svc: RosService,
    F: ClientReplyCallback<Svc::Reply>,
{
    use core::sync::atomic::Ordering;
    let entry = unsafe { &mut *(ptr as *mut ServiceClientCallbackEntry<Svc, F, REPLY_BUF>) };

    if !entry.hdr.pending {
        return Ok(false);
    }
    entry.hdr.reply_ready.store(false, Ordering::Release);

    // Issue 2009 — see `service_client_raw_try_process`.
    let ServiceClientSendHeader {
        handle,
        reply_buffer,
        sequences,
        ..
    } = &mut entry.hdr;
    match take_client_reply(handle, reply_buffer, sequences) {
        Ok(Some((len, _seq))) => {
            entry.hdr.pending = false;
            let mut reader = CdrReader::new_with_header(&entry.hdr.reply_buffer[..len])
                .map_err(|_| TransportError::DeserializationError)?;
            // Fully-qualify the `Deserialize` trait (mirrors the
            // `DeserializeView` call above): arena.rs imports
            // `DeserializeView` but not `Deserialize`, so the bare
            // `Svc::Reply::deserialize` only resolved when a default/std feature
            // happened to glob it into scope — under `rmw-cffi` (embedded) it
            // failed E0599. The fully-qualified path resolves under every feature.
            let reply = <Svc::Reply as nros_serdes::Deserialize>::deserialize(&mut reader)
                .map_err(|_| TransportError::DeserializationError)?;
            // Phase 8 — hooked AFTER both deserialization `?`s. Placing the
            // start any earlier would let a `DeserializationError` return
            // between start and end and leave an unbalanced span; here the
            // callback is the only thing between the pair, so the measured
            // interval is user code and nothing else.
            // Issue 2148 — the info is read from the handle that matched this
            // reply, before anything else can take on it (as
            // `Promise::take_with_info` does); only the with-info shape reads it,
            // and that one copy of a plain struct is inside the span.
            let handle = &entry.hdr.handle;
            trace_cb_start(desc_idx);
            entry
                .callback
                .on_reply(&reply, || nros_rmw::ClientTrait::response_info(handle));
            trace_cb_end(desc_idx);
            Ok(true)
        }
        Ok(None) => Ok(false),
        Err(_) => {
            entry.hdr.pending = false;
            Err(TransportError::ServiceRequestFailed)
        }
    }
}

/// Typed action-client callback entry (RFC-0041, Phase 239.2). The executor
/// eager-drains the three client receives (goal-response / feedback / result)
/// at `spin_once` and dispatches them as deserialized `A::Feedback` / `A::Result`
/// to typed closures — the typed analogue of [`ActionClientRawArenaEntry`]. The
/// send side (`send_goal` / `get_result`) goes through the embedded `core`
/// (offset 0) via an `ActionClientCallback` handle.
#[repr(C)]
pub(crate) struct ActionClientCallbackEntry<
    A: RosAction,
    GRespF,
    FbF,
    ResF,
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
> {
    pub(crate) core: ActionClientCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>,
    /// RFC-0041 / Phase 239.5 — the feedback stream's QoS-depth buffer. The
    /// callback path drains `core.feedback_subscriber` directly into this ring
    /// (depth > 1) or triple-buffer (depth ≤ 1), so a burst of feedbacks between
    /// spins is buffered/reported instead of overwriting a single slot — and the
    /// shared `ActionClientCore` buffers (the `Promise` path) stay untouched.
    pub(crate) feedback_buffer: BufferStrategy,
    pub(crate) on_goal_response: GRespF,
    pub(crate) on_feedback: FbF,
    pub(crate) on_result: ResF,
    pub(crate) _phantom: PhantomData<A>,
}

/// Decode one raw feedback slot (outer header + GoalUuid at [4..20] + inner-CDR
/// payload at `offset`) and invoke the typed feedback closure (Phase 239.5).
///
/// `desc_idx` is threaded in purely so the Phase 8 callback-trace pair can
/// live at the invocation itself. This function has no entry and no slot of
/// its own to recover an identity from, and it returns WITHOUT firing on a
/// short payload or a failed field read — so bracketing its two call sites
/// instead would report invocations that never happened.
#[inline]
fn dispatch_feedback<A, F, const FEEDBACK_BUF: usize>(
    data: &[u8],
    on_feedback: &mut F,
    desc_idx: u8,
) where
    A: RosAction,
    F: FnMut(&nros_core::GoalUuid, &A::Feedback),
{
    // Issue 2014 — offsets from `action_frame`.
    let (Some(goal_id), Some(fields)) = (
        crate::action_frame::feedback_goal_id(data),
        crate::action_frame::feedback_fields(data),
    ) else {
        return;
    };
    // #175 — restore the feedback's per-message encap if a typed transport
    // framing (Cyclone) stripped it; the message's top-level encap is `data[0..4]`.
    //
    // Phase 8 — hooked INSIDE the `if let Some(fb)`: the early `return`
    // above and a `None` from `read_action_field` both leave the user
    // closure un-run, and neither may emit a span.
    if let Some(fb) = read_action_field::<A::Feedback, FEEDBACK_BUF>(data, fields) {
        trace_cb_start(desc_idx);
        on_feedback(&goal_id, &fb);
        trace_cb_end(desc_idx);
    }
}

/// Monomorphized typed action-client dispatch (RFC-0041, Phase 239.2). Mirrors
/// [`action_client_raw_try_process`] but deserializes the feedback / result
/// payloads into `A::Feedback` / `A::Result` and invokes the typed closures.
///
/// # Safety
/// `ptr` must point to a valid, aligned `ActionClientCallbackEntry<…>`.
#[allow(clippy::type_complexity)]
pub(crate) unsafe fn action_client_callback_try_process<
    A,
    GRespF,
    FbF,
    ResF,
    const GOAL_BUF: usize,
    const RESULT_BUF: usize,
    const FEEDBACK_BUF: usize,
>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    A: RosAction,
    GRespF: FnMut(&nros_core::GoalUuid, bool),
    FbF: FnMut(&nros_core::GoalUuid, &A::Feedback),
    ResF: FnMut(&nros_core::GoalUuid, nros_core::GoalStatusCode, &A::Result),
{
    let entry = unsafe {
        &mut *(ptr as *mut ActionClientCallbackEntry<
            A,
            GRespF,
            FbF,
            ResF,
            GOAL_BUF,
            RESULT_BUF,
            FEEDBACK_BUF,
        >)
    };
    let ActionClientCallbackEntry {
        core,
        feedback_buffer,
        on_goal_response,
        on_feedback,
        on_result,
        _phantom,
    } = entry;

    let mut did_work = false;

    // 1. Goal-acceptance reply (single-outstanding → gated single buffer).
    // Issue 2009 — the goal is the one the reply's request was ABOUT, not the
    // most recently sent one.
    if let Ok(Some((total_len, goal_id))) = core.try_recv_send_goal_reply_for() {
        let accepted = crate::action_frame::goal_accepted(&core.result_buffer[..total_len]);
        // Phase 8 — hooked inside the `Ok(Some(..))` arm; an empty poll is
        // the common case and runs no user code.
        trace_cb_start(desc_idx);
        on_goal_response(&goal_id, accepted);
        trace_cb_end(desc_idx);
        did_work = true;
    }

    // 2. Feedback — a stream: drain `feedback_subscriber` into the QoS-depth ring
    //    (Phase 239.5), then dispatch each slot. Each slot holds the raw feedback
    //    message: outer CDR header (4) + GoalUuid (16) + inner-CDR payload; see the
    //    raw dispatcher for the layout rationale (233.6).
    {
        match feedback_buffer {
            BufferStrategy::Triple(tb) => {
                let slot = tb.write_slot();
                if let Ok(Some(len)) = core.feedback_subscriber.take_serialized(slot) {
                    tb.writer_publish(len);
                }
                if let Some((data, len)) = tb.reader_acquire() {
                    // Phase 8 — the pair lives INSIDE `dispatch_feedback`
                    // (it can return without firing); `desc_idx` is threaded
                    // in so the leaf can name itself.
                    dispatch_feedback::<A, _, FEEDBACK_BUF>(&data[..len], on_feedback, desc_idx);
                    did_work = true;
                }
            }
            BufferStrategy::Ring(ring) => {
                while let Some(slot) = ring.try_push() {
                    match core.feedback_subscriber.take_serialized(slot) {
                        Ok(Some(len)) => ring.commit_push(len),
                        _ => break,
                    }
                }
                while let Some((data, len)) = ring.try_pop() {
                    // Phase 8 — one pair per drained slot, emitted by
                    // `dispatch_feedback` itself, so a ring draining N
                    // messages reports N invocations rather than one.
                    dispatch_feedback::<A, _, FEEDBACK_BUF>(&data[..len], on_feedback, desc_idx);
                    ring.commit_pop();
                    did_work = true;
                }
            }
        }
    }

    // 3. Result reply — status at byte 4, payload at [8 ..] (header + status +
    //    align pad); see the raw dispatcher (Phase 96.1).
    if let Ok(Some((total_len, goal_id))) = core.try_recv_get_result_reply_for() {
        let reply = &core.result_buffer[..total_len];
        // Issue 2014 — status and payload offset from `action_frame`.
        if let (Some(status), Some(fields)) = (
            crate::action_frame::result_status(reply),
            crate::action_frame::result_fields(reply),
        ) {
            // #175 — restore the result's per-message encap if a typed transport
            // framing (Cyclone) stripped it; the reply's top-level encap is
            // `reply[0..4]`.
            if let Some(res) = read_action_field::<A::Result, RESULT_BUF>(reply, fields) {
                // Phase 8 — inside the `Some(res)` arm: a failed
                // deserialize leaves `on_result` un-run and must emit
                // nothing. Distinct callback, its own pair.
                trace_cb_start(desc_idx);
                on_result(&goal_id, status, &res);
                trace_cb_end(desc_idx);
            }
        }
        did_work = true;
    }

    Ok(did_work)
}

/// Issue 2081 — how a raw service entry calls its callback: with the request
/// bytes alone, or with the request's id beside them.
pub(crate) trait RawServiceDispatch: Copy {
    /// Call the callback for one taken request. `handle` is the server the
    /// request was just taken from, so its `request_info()` describes it.
    ///
    /// # Safety
    /// The pointers are the dispatch's own buffers, valid for the call.
    #[allow(clippy::too_many_arguments)]
    unsafe fn invoke(
        self,
        handle: &session::RmwServiceServer,
        req: *const u8,
        req_len: usize,
        resp: *mut u8,
        resp_cap: usize,
        resp_len: *mut usize,
        context: *mut core::ffi::c_void,
    ) -> bool;
}

impl RawServiceDispatch for RawServiceCallback {
    unsafe fn invoke(
        self,
        _handle: &session::RmwServiceServer,
        req: *const u8,
        req_len: usize,
        resp: *mut u8,
        resp_cap: usize,
        resp_len: *mut usize,
        context: *mut core::ffi::c_void,
    ) -> bool {
        unsafe { self(req, req_len, resp, resp_cap, resp_len, context) }
    }
}

impl RawServiceDispatch for RawServiceRequestIdCallback {
    unsafe fn invoke(
        self,
        handle: &session::RmwServiceServer,
        req: *const u8,
        req_len: usize,
        resp: *mut u8,
        resp_cap: usize,
        resp_len: *mut usize,
        context: *mut core::ffi::c_void,
    ) -> bool {
        // The take that just succeeded is the one `request_info` describes.
        let mut id = RawRequestId::from_service_info(&nros_rmw::ServiceTrait::request_info(handle));
        unsafe { self(req, req_len, &mut id, resp, resp_cap, resp_len, context) }
    }
}

/// Monomorphized raw service dispatch function.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SrvRawEntry<REQ_BUF, REPLY_BUF, C>`.
pub(crate) unsafe fn srv_raw_try_process<
    const REQ_BUF: usize,
    const REPLY_BUF: usize,
    C: RawServiceDispatch,
>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError> {
    let entry = unsafe { &mut *(ptr as *mut SrvRawEntry<REQ_BUF, REPLY_BUF, C>) };
    let SrvRawEntry {
        handle,
        req_buffer,
        reply_buffer,
        callback,
        context,
        capture: _,
    } = entry;
    let buf_start = req_buffer.as_ptr() as usize;
    let (data_offset, data_len, seq_num) = match handle.take_request(req_buffer) {
        Ok(Some(request)) => {
            let offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
            let len = request.data.len();
            let seq = request.sequence_number;
            (offset, len, seq)
        }
        Ok(None) => return Ok(false),
        // issue 1088 — a request is pending and the backend has no free
        // reply-correlation slot. Keep the KIND: rewriting it to
        // `ServiceReplyFailed` made a saturated server read as a broken one in
        // `SpinOnceResult`, and hid the one condition the caller can act on.
        Err(TransportError::WouldBlock) => return Err(TransportError::WouldBlock),
        Err(_) => return Err(TransportError::ServiceReplyFailed),
    };

    let mut resp_len: usize = 0;
    // Phase 8 — hooked below the `Ok(None) => return Ok(false)` arm of the
    // receive, so an empty poll emits nothing. The `?` on `send_response`
    // sits AFTER `trace_cb_end`, so no early exit can strand an open span.
    trace_cb_start(desc_idx);
    let ok = unsafe {
        callback.invoke(
            handle,
            req_buffer.as_ptr().add(data_offset),
            data_len,
            reply_buffer.as_mut_ptr(),
            REPLY_BUF,
            &mut resp_len,
            *context,
        )
    };
    trace_cb_end(desc_idx);
    if ok && resp_len > 0 {
        handle
            .send_response(seq_num, &reply_buffer[..resp_len])
            .map_err(|_| TransportError::ServiceReplyFailed)?;
    }
    Ok(true)
}

/// Monomorphized guard condition dispatch function.
///
/// # Safety
/// `ptr` must point to a valid, aligned `GuardConditionEntry<F>`.
pub(crate) unsafe fn guard_try_process<F>(
    ptr: *mut u8,
    _delta_us: u64,
    desc_idx: u8,
) -> Result<bool, TransportError>
where
    F: FnMut(),
{
    let entry = unsafe { &mut *(ptr as *mut GuardConditionEntry<F>) };
    if entry.flag.swap(false, portable_atomic::Ordering::AcqRel) {
        // Phase 8 — hooked inside the flag-consuming branch, the same shape
        // as `timer_try_process`. A guard is polled on every spin and is
        // almost always un-triggered, so `Ok(false)` is the common outcome
        // and must stay silent.
        trace_cb_start(desc_idx);
        (entry.callback)();
        trace_cb_end(desc_idx);
        Ok(true)
    } else {
        Ok(false)
    }
}

// ============================================================================
// Readiness check functions
// ============================================================================

/// SubInfoEntry readiness.
///
/// # Safety
/// `ptr` must point to a valid `SubInfoEntry<M, F>`.
pub(crate) unsafe fn sub_info_has_data<M, F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubInfoEntry<M, F>) };
    entry.handle.has_data()
}

/// SubSafetyEntry readiness.
///
/// # Safety
/// `ptr` must point to a valid `SubSafetyEntry<M, F>`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_safety_has_data<M, F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const SubSafetyEntry<M, F>) };
    entry.handle.has_data()
}

/// Service readiness: check `has_request()` on the service handle.
///
/// # Safety
/// `ptr` must point to a valid `SrvEntry<Svc, F, RQ, RP>`.
pub(crate) unsafe fn srv_has_data<Svc: RosService, F, const RQ: usize, const RP: usize>(
    ptr: *const u8,
) -> bool {
    let entry = unsafe { &*(ptr as *const SrvEntry<Svc, F, RQ, RP>) };
    entry.handle.has_request()
}

/// Raw service readiness.
///
/// # Safety
/// `ptr` must point to a valid `SrvRawEntry<RQ, RP, C>`.
pub(crate) unsafe fn srv_raw_has_data<const RQ: usize, const RP: usize, C: RawServiceDispatch>(
    ptr: *const u8,
) -> bool {
    let entry = unsafe { &*(ptr as *const SrvRawEntry<RQ, RP, C>) };
    entry.handle.has_request()
}

/// Guard condition readiness: check the atomic flag.
///
/// # Safety
/// `ptr` must point to a valid `GuardConditionEntry<F>`.
pub(crate) unsafe fn guard_has_data<F>(ptr: *const u8) -> bool {
    let entry = unsafe { &*(ptr as *const GuardConditionEntry<F>) };
    entry.flag.load(portable_atomic::Ordering::Acquire)
}

/// Timers and action entries are always considered ready.
pub(crate) unsafe fn always_ready(_ptr: *const u8) -> bool {
    true
}

// ============================================================================
// LET pre-sample functions
// ============================================================================

/// Pre-sample a typed subscription with MessageInfo for LET mode.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubInfoEntry<M, F>`.
pub(crate) unsafe fn sub_info_pre_sample<M, F>(ptr: *mut u8) {
    let entry = unsafe { &mut *(ptr as *mut SubInfoEntry<M, F>) };
    // For LET, we sample only the data (MessageInfo is not preserved in the snapshot)
    entry.sampled_len = match entry.handle.take_serialized(entry.buffer.as_mut_slice()) {
        Ok(Some(len)) => len,
        _ => 0,
    };
}

/// Pre-sample a safety subscription for LET mode.
///
/// # Safety
/// `ptr` must point to a valid, aligned `SubSafetyEntry<M, F>`.
#[cfg(feature = "safety-e2e")]
pub(crate) unsafe fn sub_safety_pre_sample<M, F>(ptr: *mut u8) {
    let entry = unsafe { &mut *(ptr as *mut SubSafetyEntry<M, F>) };
    entry.sampled_len = match entry.handle.take_serialized(entry.buffer.as_mut_slice()) {
        Ok(Some(len)) => len,
        _ => 0,
    };
}

/// No-op pre-sample for non-subscription entries (services, timers, etc.).
pub(crate) unsafe fn no_pre_sample(_ptr: *mut u8) {}

// ============================================================================
// Monomorphized handle operation functions
// ============================================================================

/// Action server: publish feedback via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_publish_feedback<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
    goal_id: &nros_core::GoalUuid,
    feedback: &A::Feedback,
) -> Result<(), NodeError>
where
    A: RosAction,
{
    let entry =
        unsafe { &mut *(ptr as *mut ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    entry.server.publish_feedback(goal_id, feedback)
}

/// Action server: complete a goal via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_complete_goal<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
    goal_id: &nros_core::GoalUuid,
    status: nros_core::GoalStatusCode,
    result: A::Result,
) -> Result<(), NodeError>
where
    A: RosAction,
{
    let entry =
        unsafe { &mut *(ptr as *mut ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    entry.server.complete_goal(goal_id, status, result)
}

/// Action server: set goal status via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_set_goal_status<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
    goal_id: &nros_core::GoalUuid,
    status: nros_core::GoalStatusCode,
) where
    A: RosAction,
{
    let entry =
        unsafe { &mut *(ptr as *mut ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    entry.server.set_goal_status(goal_id, status);
}

/// Action server: get active goal count via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_active_goal_count<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *const u8,
) -> usize
where
    A: RosAction,
{
    let entry =
        unsafe { &*(ptr as *const ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    entry.server.active_goal_count()
}

/// Action server: eagerly reclaim delivered results — phase-417 W4.b.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_expire_goals<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
) -> usize
where
    A: RosAction,
{
    let entry =
        unsafe { &mut *(ptr as *mut ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    entry.server.expire_goals()
}

/// Action server: does this server still know `goal_id`? — phase-417 W4.b.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_goal_exists<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *const u8,
    goal_id: &nros_core::GoalUuid,
) -> bool
where
    A: RosAction,
{
    let entry =
        unsafe { &*(ptr as *const ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    entry.server.goal_exists(goal_id)
}

/// Raw action server: publish feedback via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_publish_feedback<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
    goal_id: &nros_core::GoalUuid,
    feedback_data: *const u8,
    feedback_len: usize,
) -> Result<(), NodeError> {
    let entry = unsafe { &mut *(ptr as *mut ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    let feedback_cdr = unsafe { core::slice::from_raw_parts(feedback_data, feedback_len) };
    entry.core.publish_feedback_raw(goal_id, feedback_cdr)
}

/// Raw action server: complete a goal via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_complete_goal<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
    goal_id: &nros_core::GoalUuid,
    status: nros_core::GoalStatusCode,
    result_data: *const u8,
    result_len: usize,
) -> Result<(), NodeError> {
    let entry = unsafe { &mut *(ptr as *mut ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    let result_cdr = unsafe { core::slice::from_raw_parts(result_data, result_len) };
    entry.core.complete_goal_raw(goal_id, status, result_cdr)
}

/// Raw action server: set goal status via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_set_goal_status<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
    goal_id: &nros_core::GoalUuid,
    status: nros_core::GoalStatusCode,
) {
    let entry = unsafe { &mut *(ptr as *mut ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    entry.core.set_goal_status(goal_id, status);
}

/// Raw action server: get active goal count via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_active_goal_count<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *const u8,
) -> usize {
    let entry = unsafe { &*(ptr as *const ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    entry.core.active_goal_count()
}

/// Raw action server: eagerly reclaim delivered results — phase-417 W4.b.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_expire_goals<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *mut u8,
) -> usize {
    let entry = unsafe { &mut *(ptr as *mut ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    entry.core.expire_completed_results()
}

/// Raw action server: does this server still know `goal_id`? — phase-417 W4.b.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_goal_exists<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *const u8,
    goal_id: &nros_core::GoalUuid,
) -> bool {
    let entry = unsafe { &*(ptr as *const ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    entry.core.goal_exists(goal_id)
}

/// Raw action server: iterate active goals via arena entry.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerRawArenaEntry`.
pub(crate) unsafe fn as_raw_for_each_active_goal<
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *const u8,
    f: &mut dyn FnMut(&super::action_core::RawActiveGoal),
) {
    let entry = unsafe { &*(ptr as *const ActionServerRawArenaEntry<GB, RB, FB, MG>) };
    for goal in entry.core.active_goals() {
        f(goal);
    }
}

/// Action server: iterate active goals via arena entry.
///
/// Calls `f` for each active goal, reconstructing `ActiveGoal<A>` from
/// the core's `RawActiveGoal` and the parallel typed goals vec.
///
/// # Safety
/// `ptr` must point to a valid `ActionServerArenaEntry`.
pub(crate) unsafe fn as_for_each_active_goal<
    A,
    GoalF,
    CancelF,
    const GB: usize,
    const RB: usize,
    const FB: usize,
    const MG: usize,
>(
    ptr: *const u8,
    f: &mut dyn FnMut(&ActiveGoal<A>),
) where
    A: RosAction + 'static,
    A::Goal: Clone,
{
    let entry =
        unsafe { &*(ptr as *const ActionServerArenaEntry<A, GoalF, CancelF, GB, RB, FB, MG>) };
    for (i, raw_goal) in entry.server.core.active_goals().iter().enumerate() {
        let active = ActiveGoal {
            goal_id: raw_goal.goal_id,
            status: raw_goal.status,
            goal: entry.server.typed_goals[i].clone(),
        };
        f(&active);
    }
}

#[cfg(test)]
mod borrowed_sub_tests {
    use nros_core::{CdrReader, CdrWriter, DeserError};

    use super::*;

    // Hand-written borrowed message mirroring what codegen will emit for
    // `{ uint32 width; uint8[] data; }` in `borrowed` mode (Phase 229.6).
    struct ImageView<'a> {
        width: u32,
        data: &'a [u8],
    }

    impl<'a> DeserializeView<'a> for ImageView<'a> {
        fn deserialize_view(reader: &mut CdrReader<'a>) -> Result<Self, DeserError> {
            let width = reader.read_u32()?;
            let data = reader.read_slice_u8()?;
            Ok(ImageView { width, data })
        }
    }

    // Zero-sized borrowed-family marker (codegen emits `struct ImageViewable;`).
    struct ImageViewable;
    impl ViewableMessage for ImageViewable {
        type View<'a> = ImageView<'a>;
        const TYPE_NAME: &'static str = "test_msgs::msg::dds_::Image_";
        const TYPE_HASH: &'static str = "borrowed-test-hash";
    }

    // The borrowed view must alias the source CDR buffer (no `heapless::Vec`
    // copy) — the whole point of `borrowed` mode (issue 0007).
    #[test]
    fn borrowed_view_is_zero_copy_into_source_buffer() {
        let payload: [u8; 64] = core::array::from_fn(|i| i as u8);
        let mut buf = [0u8; 128];
        let written = {
            let mut w = CdrWriter::new_with_header(&mut buf).unwrap();
            w.write_u32(7).unwrap();
            w.write_sequence_len(payload.len()).unwrap();
            w.write_bytes(&payload).unwrap();
            w.position()
        };

        let mut reader = CdrReader::new_with_header(&buf[..written]).unwrap();
        let view = ImageView::deserialize_view(&mut reader).unwrap();

        assert_eq!(view.width, 7);
        assert_eq!(view.data, &payload[..]);

        // The borrowed slice points INTO `buf`, proving zero-copy.
        let buf_start = buf.as_ptr() as usize;
        let buf_end = buf_start + buf.len();
        let data_ptr = view.data.as_ptr() as usize;
        assert!(
            data_ptr >= buf_start && data_ptr < buf_end,
            "borrowed data must alias the source buffer (zero-copy)"
        );
    }

    // Phase 231 Wave 3 (RFC-0038) — single-copy proof. The in-place subscription
    // entry carries handle + callback only; the buffered entry additionally
    // carries the arena `BufferStrategy` (the copy-#1 staging buffer). So the
    // in-place entry is strictly smaller — proving the arena buffer (and copy #1)
    // is gone for backends that support in-place dispatch.
    #[test]
    fn inplace_entry_drops_the_arena_buffer() {
        type Cb = fn(&u32);
        assert!(
            core::mem::size_of::<SubInplaceEntry<u32, Cb>>()
                < core::mem::size_of::<SubBufferedEntry<u32, Cb>>(),
            "in-place entry must be smaller than the buffered entry (no arena BufferStrategy)"
        );
    }

    // Compile-time proof that the codegen marker + GAT + a borrowed closure
    // satisfy exactly the bounds the executor's borrowed dispatch
    // (`sub_buffered_view_try_process`) and registration require.
    fn assert_borrowed_sub_bounds<B, F>(_callback: F)
    where
        B: ViewableMessage + 'static,
        F: for<'a> FnMut(&B::View<'a>) + 'static,
    {
    }

    #[test]
    fn borrowed_marker_satisfies_dispatch_bounds() {
        assert_borrowed_sub_bounds::<ImageViewable, _>(|view: &ImageView<'_>| {
            let _ = view.width;
            let _ = view.data.len();
        });
    }
}

/// RFC-0069 / issues 0418 + 0035 — the action payload envelope carries exactly
/// ONE CDR header, and reading it back must not eat a data word.
///
/// The RFC lists this as the acceptance item "most likely to be skipped: its
/// absence is why the divergence survived a redesign in the first place." These
/// sit at the level the corruption happens — `read_action_field`, the consumer
/// half of the 0418 producer/consumer pair.
#[cfg(test)]
mod action_envelope_tests {
    use nros_core::{CdrWriter, Deserialize, Serialize};

    use super::*;

    /// Three `int32`s. The FIRST is the one that matters: give it the value 256
    /// and the body's leading bytes are `00 01 00 00` — byte for byte the
    /// little-endian CDR encapsulation header.
    #[derive(Debug, PartialEq)]
    struct Body {
        first: i32,
        rest: [i32; 2],
    }

    impl Serialize for Body {
        fn serialize(&self, w: &mut CdrWriter) -> Result<(), nros_serdes::SerError> {
            w.write_i32(self.first)?;
            for v in &self.rest {
                w.write_i32(*v)?;
            }
            Ok(())
        }
    }

    impl Deserialize for Body {
        fn deserialize(r: &mut CdrReader) -> Result<Self, nros_serdes::DeserError> {
            Ok(Body {
                first: r.read_i32()?,
                rest: [r.read_i32()?, r.read_i32()?],
            })
        }
    }

    const LE_ENCAP: [u8; 4] = [0x00, 0x01, 0x00, 0x00];

    /// Serialize the way the 0418 PRODUCER does: fields only, no inner header.
    fn produce_headerless(body: &Body, buf: &mut [u8]) -> usize {
        let mut w = CdrWriter::new(buf);
        body.serialize(&mut w).expect("serialize");
        w.position()
    }

    #[test]
    fn headerless_payload_round_trips() {
        let body = Body {
            first: 7,
            rest: [1, 1],
        };
        let mut buf = [0u8; 64];
        let n = produce_headerless(&body, &mut buf);

        let got = read_action_field::<Body, 64>(&LE_ENCAP, &buf[..n])
            .expect("consumer must decode the headerless payload");
        assert_eq!(got, body, "issue 0035: the reader ate a data word");
    }

    /// THE hazard 0418's own doc names and did not test.
    ///
    /// `payload_has_cdr_encap` was a VALUE sniff: a leading `int32` of 256 is
    /// `00 01 00 00`, indistinguishable from the LE encap header. While the
    /// producer wrote an inner header only Cyclone took the other branch, so it
    /// never mattered; once 0418 made every payload headerless the sniff was
    /// consulted for all of them, and this body had its first word eaten as
    /// framing — `first` came back as `rest[0]`. Issue #35 reached through a
    /// payload VALUE rather than a framing bug.
    #[test]
    fn a_leading_word_that_looks_like_an_encap_is_data_not_framing() {
        let body = Body {
            first: 256,
            rest: [11, 12],
        };
        let mut buf = [0u8; 64];
        let n = produce_headerless(&body, &mut buf);

        assert_eq!(
            &buf[..4],
            &LE_ENCAP,
            "precondition: an int32 of 256 IS the LE encap byte pattern"
        );
        assert!(
            payload_has_cdr_encap(&buf[..n]),
            "precondition: the retired sniff cannot tell this from a header"
        );

        let got = read_action_field::<Body, 64>(&LE_ENCAP, &buf[..n]).expect("must decode");
        assert_eq!(
            got, body,
            "the leading word is DATA — reading it as framing is issue #35"
        );
    }

    /// The version break RFC-0069 accepts, asserted rather than assumed.
    ///
    /// A pre-0418 peer sends `[inner header][fields]`. The consumer no longer
    /// sniffs, so it splices unconditionally and the inner header is read as
    /// data. Old and new images are wire-incompatible on action payloads — the
    /// RFC's "silent version skew" risk, made explicit here so the next reader
    /// finds it as a decision and not as a mystery.
    #[test]
    fn a_pre_0418_double_header_payload_does_not_decode_as_itself() {
        let body = Body {
            first: 5,
            rest: [6, 7],
        };
        let mut inner = [0u8; 64];
        let n = produce_headerless(&body, &mut inner);
        let mut withhdr = [0u8; 68];
        withhdr[..4].copy_from_slice(&LE_ENCAP);
        withhdr[4..4 + n].copy_from_slice(&inner[..n]);

        let got = read_action_field::<Body, 68>(&LE_ENCAP, &withhdr[..4 + n]);
        assert_ne!(
            got,
            Some(body),
            "a pre-0418 payload must NOT silently appear to decode — the wire \
             format changed and the skew is expected to fail"
        );
    }
}

/// Issue #505 — timer overrun policy. `timer_try_process` takes the
/// elapsed delta directly, so a stall is expressible as a single large
/// delta and every case here is a pure unit test.
#[cfg(test)]
mod timer_overrun_tests {
    use super::*;

    /// Drive `entry` through one `try_process` pass, crediting `delta_ms`
    /// worth of elapsed time (the dispatcher's unit is microseconds).
    fn step<F: FnMut(&TimerCallInfo)>(entry: &mut TimerEntry<F>, delta_ms: u64) -> bool {
        // SAFETY: `entry` is a live, aligned `TimerEntry<F>`.
        unsafe {
            timer_try_process::<F>(
                (entry as *mut TimerEntry<F>).cast::<u8>(),
                delta_ms * 1000,
                0,
            )
            .unwrap()
        }
    }

    /// Same, in microseconds — the dispatcher's native unit.
    fn step_us<F: FnMut(&TimerCallInfo)>(entry: &mut TimerEntry<F>, delta_us: u64) -> bool {
        // SAFETY: `entry` is a live, aligned `TimerEntry<F>`.
        unsafe {
            timer_try_process::<F>((entry as *mut TimerEntry<F>).cast::<u8>(), delta_us, 0).unwrap()
        }
    }

    fn periodic(
        period_us: u64,
        policy: TimerOverrunPolicy,
    ) -> TimerEntry<impl FnMut(&TimerCallInfo)> {
        TimerEntry {
            period_us,
            elapsed_us: 0,
            overruns: 0,
            overruns_reported: 0,
            oneshot: false,
            fired: false,
            cancelled: false,
            overrun_policy: policy,
            clock_source: TimerClockSource::Steady,
            last_clock_ns: 0,
            callback: |_: &TimerCallInfo| {},
        }
    }

    #[test]
    fn skip_is_the_default_policy() {
        assert_eq!(TimerOverrunPolicy::default(), TimerOverrunPolicy::Skip);
    }

    #[test]
    fn skip_coalesces_a_stall_into_one_activation() {
        // The observed failure: a ~200 ms preemption of a 10 ms timer
        // replayed as a burst of activations. Under Skip the backlog
        // costs exactly one activation plus a counter bump.
        let mut t = periodic(10_000, TimerOverrunPolicy::Skip);
        assert!(step(&mut t, 205));
        assert_eq!(t.overruns, 19, "205 ms of a 10 ms period = 20 due, 1 fired");
        // Nothing is owed: the next pass waits out the remaining period.
        assert!(!step(&mut t, 4));
        assert!(step(&mut t, 1));
    }

    #[test]
    fn skip_preserves_the_phase_grid() {
        // 205 ms leaves a 5 ms remainder; keeping it means the next fire
        // lands 5 ms later, back on the original 10 ms grid, instead of
        // re-anchoring to the moment the stall ended.
        let mut t = periodic(10_000, TimerOverrunPolicy::Skip);
        assert!(step(&mut t, 205));
        assert_eq!(t.elapsed_us, 5_000);
    }

    #[test]
    fn catchup_replays_every_missed_period() {
        let mut t = periodic(10_000, TimerOverrunPolicy::CatchUp);
        assert!(step(&mut t, 205));
        // One activation per pass until the backlog drains, with no
        // further time credited.
        let mut replays = 0;
        while step(&mut t, 0) {
            replays += 1;
            assert!(replays < 64, "backlog must drain");
        }
        assert_eq!(replays, 19);
        assert_eq!(t.overruns, 0, "CatchUp loses nothing, so counts nothing");
    }

    #[test]
    fn on_time_ticks_never_count_as_overruns() {
        let mut t = periodic(10_000, TimerOverrunPolicy::Skip);
        for _ in 0..100 {
            assert!(step(&mut t, 10));
        }
        assert_eq!(t.overruns, 0);
        assert_eq!(t.elapsed_us, 0);
    }

    #[test]
    fn a_single_late_period_is_one_overrun() {
        // Exactly two periods due = one fired, one dropped.
        let mut t = periodic(10_000, TimerOverrunPolicy::Skip);
        assert!(step(&mut t, 20));
        assert_eq!(t.overruns, 1);
    }

    #[test]
    fn oneshot_ignores_the_policy() {
        let mut t = TimerEntry {
            period_us: 10_000,
            elapsed_us: 0,
            overruns: 0,
            overruns_reported: 0,
            oneshot: true,
            fired: false,
            cancelled: false,
            overrun_policy: TimerOverrunPolicy::Skip,
            clock_source: TimerClockSource::Steady,
            last_clock_ns: 0,
            callback: |_: &TimerCallInfo| {},
        };
        assert!(step(&mut t, 205));
        assert_eq!(t.overruns, 0);
        assert!(!step(&mut t, 205), "a fired one-shot stays fired");
    }

    #[test]
    fn sub_millisecond_periods_are_expressible() {
        // Issue #505 — the dispatcher's unit is microseconds, so a
        // 500 us period is a real period rather than the "fires every
        // spin" degenerate case a millisecond field forced it into.
        let mut t = periodic(500, TimerOverrunPolicy::Skip);
        // 200 us of credit is not enough...
        assert!(!step_us(&mut t, 200));
        // ...600 us total is.
        assert!(step_us(&mut t, 400));
        assert_eq!(t.overruns, 0);
        assert_eq!(t.elapsed_us, 100, "phase remainder survives");
    }

    #[test]
    fn a_stalled_sub_millisecond_timer_counts_whole_periods() {
        let mut t = periodic(500, TimerOverrunPolicy::Skip);
        assert!(step_us(&mut t, 5_000)); // 10 periods due
        assert_eq!(t.overruns, 9);
    }

    #[test]
    fn zero_period_does_not_divide_by_zero() {
        // Degenerate but reachable: `TimerDuration::from_millis(0)` or a
        // sub-millisecond period truncated to 0 by `as_millis`.
        let mut t = periodic(0, TimerOverrunPolicy::Skip);
        assert!(step(&mut t, 5));
        assert_eq!(t.overruns, 0);
        assert_eq!(t.elapsed_us, 0);
    }

    #[test]
    fn overrun_count_saturates_instead_of_wrapping() {
        let mut t = periodic(1_000, TimerOverrunPolicy::Skip);
        t.overruns = u32::MAX - 1;
        assert!(step(&mut t, 1_000));
        assert_eq!(t.overruns, u32::MAX);
    }
}

/// issue 1437 / phase-444 — the layout invariant the three
/// `Executor::*_handle` accessors read the arena through.
///
/// Those accessors resolve an arena entry to its transport handle from
/// `(EntryKind, offset)` ALONE, with no knowledge of the shape the entry was
/// monomorphised into — 12 subscription shapes, 2 service-server, 2
/// service-client, differing in buffer sizes, message types and callback
/// types. That works because every one of them is `#[repr(C)]` with its
/// handle as the FIRST field, so `arena_base + offset` is the handle pointer
/// for all of them.
///
/// It is an invariant nothing else states and a compiler would not catch:
/// prepending a field to any entry below leaves every caller compiling and
/// makes the accessors read the new field AS a handle. So it is measured
/// here, per shape, rather than asserted in a comment beside the cast.
#[cfg(test)]
mod handle_offset_tests {
    use super::*;

    /// One instantiation per SHAPE. The const-generic buffer sizes are
    /// deliberately not the defaults — a shape whose handle offset depended
    /// on them would be exactly the bug this exists to catch, and equal
    /// numbers everywhere would hide it.
    macro_rules! handle_at_zero {
        ($($name:literal => $ty:ty),* $(,)?) => {
            $(assert_eq!(
                core::mem::offset_of!($ty, handle),
                0,
                concat!($name, " must begin with its handle — \
                         Executor::*_handle resolves an arena entry by \
                         (kind, offset) and casts arena_base + offset \
                         straight to the handle type"),
            );)*
        };
    }

    #[test]
    fn every_arena_entry_begins_with_its_handle() {
        type Cb = fn();
        handle_at_zero! {
            "SubInfoEntry"            => SubInfoEntry<u8, Cb>,
            "SubBufferedEntry"        => SubBufferedEntry<u8, Cb>,
            "SubInplaceEntry"         => SubInplaceEntry<u8, Cb>,
            "SubBufferedRawEntry"     => SubBufferedRawEntry<Cb>,
            "SubBufferedViewEntry"    => SubBufferedViewEntry<u8, Cb>,
            "SubBufferedRawInfoEntry" => SubBufferedRawInfoEntry<Cb>,
            "SubBufferedRawInfoCEntry" => SubBufferedRawInfoCEntry,
            "SubBufferedRawCEntry"    => SubBufferedRawCEntry,
            "SubBufferedTypedCEntry"  => SubBufferedTypedCEntry,
            "SrvRawEntry"             => SrvRawEntry<43, 47>,
            "ServiceClientRawArenaEntry" => ServiceClientRawArenaEntry<53>,
            "ServiceClientSendHeader" => ServiceClientSendHeader<59>,
        }
    }
}

/// Issue 1632 — the Rust arena's drop log prints the size the sample needed
/// beside the buffer, as the C++ one has since issue 1612.
///
/// The sink keeps what was logged, so the test reads what a human would see.
/// The sink is process-global, so the cells serialise; the drop counter is
/// never touched, since other cells advance it too (issue 2082).
#[cfg(test)]
mod dropped_take_log_tests {
    use super::{drop_is_reported, log_dropped_take};
    use nros_rmw::TransportError;

    struct CapturingSink {
        lines: std::sync::Mutex<std::vec::Vec<std::string::String>>,
    }

    impl nros_log::LogSink for CapturingSink {
        fn log(&self, record: &nros_log::Record<'_>) {
            if record.message.contains("subscription take DROPPED") {
                self.lines
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .push(record.message.into());
            }
        }
    }

    static CAPTURE: std::sync::OnceLock<&'static CapturingSink> = std::sync::OnceLock::new();
    static SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(());

    /// The line this call logged. The sink is process-global, so another
    /// cell's drop may land beside it; `cap` picks this call's line out, and
    /// each call site passes a buffer size no other cell uses (issue 2082).
    fn logged_one(err: TransportError, cap: usize, recorded: Option<usize>) -> std::string::String {
        let _g = SERIAL
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let sink = CAPTURE.get_or_init(|| {
            let sink: &'static CapturingSink =
                std::boxed::Box::leak(std::boxed::Box::new(CapturingSink {
                    lines: std::sync::Mutex::new(std::vec::Vec::new()),
                }));
            assert!(nros_log::add_sink(sink), "no sink slot left");
            sink
        });
        let marker = std::format!("{cap}-byte buffer");
        let mine = || -> std::vec::Vec<std::string::String> {
            sink.lines
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .iter()
                .filter(|l| l.contains(&marker))
                .cloned()
                .collect()
        };
        sink.lines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clear();
        // A record nros_log lost to its hold queue (issue 2000) is raised
        // again. Always as drop number one, never through DROPPED_TAKES.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        let seen = nros_log::early::emit_until_delivered(
            || log_dropped_take(&err, cap, recorded, 0),
            || !mine().is_empty(),
            || {
                std::thread::yield_now();
                std::time::Instant::now() < deadline
            },
        );
        let lines = mine();
        assert!(seen, "no line for a {cap}-byte buffer: {lines:?}");
        // A re-raise can deliver the line twice; every copy is the same drop.
        assert!(
            lines.iter().all(|l| *l == lines[0]),
            "the copies disagree: {lines:?}"
        );
        lines[0].clone()
    }

    #[test]
    fn the_drop_log_names_the_sample_size_beside_the_buffer() {
        let line = logged_one(TransportError::BufferTooSmall, 4093, Some(1000));
        assert!(line.contains("1000-byte sample"), "no sample size: {line}");
        assert!(line.contains("4093-byte buffer"), "no buffer size: {line}");
        // nros_log's buffer is all-or-nothing; the issue number is the tail.
        assert!(line.contains("(issue 0757)"), "truncated: {line}");
    }

    #[test]
    fn an_unknown_size_and_a_staging_refusal_print_as_unknown() {
        let unknown = logged_one(TransportError::BufferTooSmall, 4091, None);
        assert!(unknown.contains("?-byte sample"), "{unknown}");
        // `MessageTooLarge` is the backend's staging buffer, refused before a
        // take measured anything: an earlier take's size must not be printed.
        let staging = logged_one(TransportError::MessageTooLarge, 4087, Some(1000));
        assert!(staging.contains("?-byte sample"), "{staging}");
        assert!(!staging.contains("1000"), "{staging}");
    }

    #[test]
    fn the_first_drop_and_every_64th_are_reported() {
        assert!(drop_is_reported(0));
        assert!(!drop_is_reported(1));
        assert!(!drop_is_reported(63));
        assert!(drop_is_reported(64));
        assert!(!drop_is_reported(65));
        assert!(drop_is_reported(128));
    }
}