Skip to main content

nros_node/executor/
monitor.rs

1//! RFC-0052 / phase-296 W3b.4/.5 — on-target contract monitors.
2//!
3//! The baked shape mirrors Phase 211.H's `qos_overrides`: codegen emits a
4//! `&'static [MonitorSpec]` table (plus one `static PubMonitorCell` per
5//! contracted publisher) from the SystemModel's contract layer; the entry
6//! installs it on the executor before entity creation. An uncontracted
7//! image bakes an empty table — every path below dead-code-eliminates.
8//!
9//! Publish counting is an atomic bump on the publisher handle (no clock,
10//! no lock on the hot path); the rate check runs on spin ticks over a
11//! ~[`RATE_CHECK_INTERVAL_US`] window and pushes violations into a small
12//! ring the entry glue drains into the `nros-diagnostics` reporter.
13//!
14//! W3b.5 adds three more rules on the same drain:
15//! - `max-age-runtime` — subscriber take-age (epoch_now - header.stamp
16//!   peeked from the raw CDR buffer at `RosMessage::STAMP_OFFSET`,
17//!   recorded into a [`SubMonitorCell`] on the take path).
18//! - `max-latency-runtime` — node-path (take → publish) latency: the
19//!   dispatch elapsed time is attributed to every monitored publisher
20//!   whose counter advanced during that dispatch (an upper bound on
21//!   take → publish, measured on the executor's monotonic clock). The span
22//!   is callback entry to exit, for every dispatch kind: a timer callback
23//!   is timed whether it fires in the drains or in the trigger-miss sweep
24//!   (phase-474 I6), and a service call it makes is inside the span. The
25//!   timer's release jitter (a late start) is not part of it; that is
26//!   `release-jitter-runtime`.
27//! - `deadline-miss-runtime` — a dispatched callback ran past its bound
28//!   SchedContext's `deadline_us`; what ELSE happens is the tier's
29//!   [`DeadlineAction`](super::sched_context::DeadlineAction).
30//!
31//! phase10-B3 adds the route rule beside it:
32//! - `route-latency-runtime` -- a node path whose trigger is an INPUT is
33//!   budgeted from the TAKE of that input to its output, and on a node that
34//!   reads its input on a timer tick that span includes the wait for the next
35//!   tick. `max-latency-runtime` times one dispatch and cannot see the wait,
36//!   so a route made of a long tick wait and a slow dispatch could exceed its
37//!   budget while every callback passed. The executor stamps the dispatch that
38//!   takes the trigger input (the earliest take not yet answered) and, at the
39//!   end of a dispatch that published one of the path's outputs, judges
40//!   take-to-dispatch-end against the path's `max_latency_ms`
41//!   (`route_take`, `route_judge`). Record only: the rule has no action.
42//!
43//! phase-462 W2 adds one more on the same drain, and it is the other half of
44//! what the contract's `on_violation` lowers to:
45//! - `silence-runtime` -- a contracted subscription that took NOTHING for a
46//!   whole `max_age_ms` window (`check_silence`). A violation is either "the
47//!   callback ran too long", which is the deadline action, or "the input
48//!   stopped coming", which is this; the four rules above can only see the
49//!   first kind.
50
51use core::sync::atomic::Ordering;
52// portable-atomic: RMW ops (fetch_add/fetch_max/swap) exist even on
53// riscv32imc / Cortex-M0+ that lack native CAS (same choice as
54// `SporadicState` / `AtomicSporadicState` in sched_context.rs).
55use portable_atomic::AtomicU32;
56
57/// One contracted publisher's counters. Baked as a `static` by codegen
58/// (or declared by the fixture); the publisher handle bumps `count` on
59/// every publish, the executor reads deltas on spin ticks.
60#[derive(Debug, Default)]
61pub struct PubMonitorCell {
62    pub count: AtomicU32,
63    /// W3b.5 — max observed take→publish latency (µs) in the current
64    /// check window. Written by the dispatch loop (fetch_max), drained
65    /// (swap 0) by the latency check.
66    pub max_latency_us: AtomicU32,
67    /// Age of the stamp this publisher last put ON THE WIRE, in
68    /// microseconds: epoch_now - outgoing header.stamp.
69    ///
70    /// Distinct from `max_latency_us`, which times this node's own
71    /// take→publish work. This says how old the DATA is that the node just
72    /// published, which is the quantity a chain is made of.
73    ///
74    /// It exists to answer a question `max-age-runtime` cannot. That rule
75    /// measures epoch_now - stamp on the TAKE path, so if every node in a
76    /// chain propagates the original stamp -- the usual ROS convention, each
77    /// node copying its input's stamp to its output -- the age at the final
78    /// consumer already IS the end-to-end latency. If any node re-stamps
79    /// with `now`, the clock silently resets and the same number becomes
80    /// single-hop age instead. Same units, same magnitude, no warning.
81    ///
82    /// A publish age near zero on a node that consumes input is the
83    /// signature of re-stamping. Recording it here is what lets a chain's
84    /// provenance be checked at all, rather than assumed.
85    ///
86    /// `0` = never observed, matching the other cells: the type has no
87    /// `STAMP_OFFSET`, or no epoch source is installed.
88    pub last_publish_stamp_age_us: AtomicU32,
89    /// phase10-B3 -- the route rule's state for this output
90    /// ([`MonitorSpec::route_trigger`]): the trigger subscription's executor
91    /// slot + 1 in the low 16 bits (0 = no subscription bound, set when the
92    /// subscription registers), [`ROUTE_VIOLATED`] and [`ROUTE_PENDING`] above.
93    pub route: AtomicU32,
94    /// phase10-B3 -- when the earliest take not yet answered by a publish of
95    /// this output happened, executor clock in microseconds (wrapping). Valid
96    /// while [`ROUTE_PENDING`] is set in `route`.
97    pub route_take_us: AtomicU32,
98}
99
100/// phase10-B3 -- `PubMonitorCell::route`: a take of the trigger input is
101/// waiting for this output.
102pub const ROUTE_PENDING: u32 = 1 << 31;
103/// phase10-B3 -- `PubMonitorCell::route`: the last judged route was over its
104/// budget (report once until one is within it, as the latency rule does).
105pub const ROUTE_VIOLATED: u32 = 1 << 30;
106/// phase10-B3 -- `PubMonitorCell::route`: the bound slot + 1.
107pub const ROUTE_SLOT_MASK: u32 = 0xffff;
108
109impl PubMonitorCell {
110    pub const fn new() -> Self {
111        Self {
112            count: AtomicU32::new(0),
113            max_latency_us: AtomicU32::new(0),
114            last_publish_stamp_age_us: AtomicU32::new(0),
115            route: AtomicU32::new(0),
116            route_take_us: AtomicU32::new(0),
117        }
118    }
119}
120
121/// One contracted subscriber's take-age accumulator (W3b.5). The take
122/// path records epoch_now - header.stamp per message (fetch_max); the
123/// age check drains it (swap 0) per window.
124#[derive(Debug, Default)]
125pub struct SubMonitorCell {
126    /// Max observed take-age (ms) in the current check window.
127    pub max_age_ms: AtomicU32,
128    /// phase-462 W2 -- observations recorded since the last check, drained
129    /// (swap 0) by `check_age`. The age itself cannot answer "did anything
130    /// arrive": a window with no take and a window whose only take was
131    /// perfectly fresh both leave `max_age_ms` at 0.
132    pub takes: AtomicU32,
133    /// phase-462 W5b -- the executor slot + 1 whose readiness is this
134    /// endpoint's arrival evidence, set when the contract's `deadline` is
135    /// judged here rather than by the backend ([`admit_contract_deadline`]).
136    /// 0 = not armed: the backend judges the deadline, or none is declared.
137    pub deadline_slot: AtomicU32,
138    /// phase-462 W5b -- spins on which the armed slot had data, drained by
139    /// `check_age`. Counted for ANY message type, stamped or not, because a
140    /// deadline is a promise about arrival and says nothing about stamps.
141    pub arrivals: AtomicU32,
142}
143
144impl SubMonitorCell {
145    pub const fn new() -> Self {
146        Self {
147            max_age_ms: AtomicU32::new(0),
148            takes: AtomicU32::new(0),
149            deadline_slot: AtomicU32::new(0),
150            arrivals: AtomicU32::new(0),
151        }
152    }
153
154    /// Take-path hook: record one message's age. `stamp_us` is the
155    /// peeked `header.stamp` as µs since the UNIX epoch, `epoch_now_us`
156    /// the receive-side wall clock. A stamp from the future clamps to 0.
157    ///
158    /// phase-462 W2 -- also the silence rule's only evidence that this
159    /// endpoint is being fed. It counts what the AGE rule can see, which is
160    /// stamped takes on a type that carries a stamp: a contracted endpoint
161    /// whose messages carry no readable stamp records nothing here and is
162    /// reported silent. That is the intended verdict rather than a gap --
163    /// its `max_age_ms` cannot be judged either, and an unjudgeable promise
164    /// reported as met is the silent class this phase exists to remove.
165    pub fn observe(&self, stamp_us: u64, epoch_now_us: u64) {
166        let age_ms = (epoch_now_us.saturating_sub(stamp_us) / 1_000).min(u32::MAX as u64) as u32;
167        self.max_age_ms.fetch_max(age_ms, Ordering::Relaxed);
168        self.takes.fetch_add(1, Ordering::Relaxed);
169    }
170
171    /// Issue 2215 -- a stamped take on an image with NO epoch source: the
172    /// stamp was read (the type carries one and it is a real time), so the
173    /// take is the silence rule's evidence exactly as [`Self::observe`]'s is,
174    /// but its age cannot be computed against a clock that is not on the
175    /// stamp's epoch, so `max_age_ms` is left alone and `max-age-runtime`
176    /// judges nothing on such an image. An unstamped take is still not
177    /// evidence: nothing here counts arrivals.
178    pub fn observe_unaged(&self) {
179        self.takes.fetch_add(1, Ordering::Relaxed);
180    }
181}
182
183/// Peek `Time { i32 sec; u32 nanosec }` little-endian at `offset` in a
184/// raw CDR receive buffer (encapsulation header included) and return µs
185/// since the UNIX epoch. `None` when the buffer is too short or the
186/// stamp is pre-epoch/zero (unstamped messages never fire age monitors).
187/// Record the age of the stamp a publisher just put on the wire.
188///
189/// Called from the publish path with the encoded CDR still in hand, using the
190/// same `STAMP_OFFSET` peek the take path uses. A no-op when the type carries
191/// no stamp, when no epoch source is installed, or when the publisher is
192/// uncontracted -- the same three ways `observe_age` folds away.
193///
194/// Stores rather than accumulates: this is "how old was the last thing
195/// published", a state, not a window maximum. A chain check wants the current
196/// value, and a max would be pinned forever by one stale message at startup.
197#[inline]
198pub fn observe_publish_stamp(cell: &PubMonitorCell, raw: &[u8], offset: usize, now_us: u64) {
199    if let Some(stamp_us) = peek_stamp_us(raw, offset) {
200        let age = now_us.saturating_sub(stamp_us).min(u32::MAX as u64) as u32;
201        cell.last_publish_stamp_age_us.store(age, Ordering::Relaxed);
202    }
203}
204
205pub fn peek_stamp_us(raw: &[u8], offset: usize) -> Option<u64> {
206    let sec_b = raw.get(offset..offset + 4)?;
207    let nsec_b = raw.get(offset + 4..offset + 8)?;
208    let sec = i32::from_le_bytes([sec_b[0], sec_b[1], sec_b[2], sec_b[3]]);
209    let nsec = u32::from_le_bytes([nsec_b[0], nsec_b[1], nsec_b[2], nsec_b[3]]);
210    if sec <= 0 {
211        return None;
212    }
213    Some(sec as u64 * 1_000_000 + nsec as u64 / 1_000)
214}
215
216/// One monitored publisher endpoint.
217#[derive(Debug, Clone, Copy)]
218pub struct MonitorSpec {
219    /// Topic name EXACTLY as the node passes it to `create_publisher`
220    /// (the SystemModel's wiring carries the same resolved name).
221    pub topic: &'static str,
222    /// Endpoint ref for violation reports (`<node FQN>/<endpoint>` — the
223    /// SystemModel contract key).
224    pub fqn: &'static str,
225    /// Declared publisher guarantee, in milli-Hz (fixed point: Hz × 1000).
226    /// 0 = no rate contract on this endpoint.
227    pub min_rate_hz_milli: u32,
228    /// W3b.5 — node-path budget (ms) for paths whose OUTPUT is this
229    /// endpoint (`contracts.node_paths[..].max_latency_ms`). 0 = no
230    /// latency contract.
231    pub max_latency_ms: u32,
232    /// phase10-B3 -- the topic of the INPUT that triggers a node path whose
233    /// output is this endpoint (`contracts.node_paths[..].input`, resolved to
234    /// its wiring topic EXACTLY as the node passes it to
235    /// `create_subscription`), or `""` for none. A subscription on this topic
236    /// registered on the same executor is the route's trigger.
237    pub route_trigger: &'static str,
238    /// phase10-B3 -- that path's budget (ms), from the take of the trigger to
239    /// this output (`route-latency-runtime`). 0 = no route contract.
240    pub route_max_latency_ms: u32,
241    /// phase-503 -- the route's SUPERVISED deadline (ms) from the take of the
242    /// trigger: a take still unanswered at `take + route_deadline_ms` is
243    /// reported (`route-deadline-runtime`) and the executor's reaction hook is
244    /// called, without waiting for an output that may never come. 0 = the
245    /// route is not supervised (only the completion verdict above judges it).
246    /// Needs `route_trigger`; ignored without it.
247    pub route_deadline_ms: u32,
248    /// phase-503 -- the reaction the hook is asked to start when the deadline
249    /// passes: an id codegen assigns to the contract's `reaction` path
250    /// (1-based; [`REACTION_NONE`] = none named). Opaque to the executor.
251    pub reaction_id: u16,
252    /// phase-503 -- which mechanism the contract chose to supervise this
253    /// route: [`MECHANISM_EXECUTOR`] (this check, in the spin loop) or
254    /// [`MECHANISM_TIMER`] (an independent timer the port arms from
255    /// `Executor::route_next_due_us`). The executor checks the deadline either
256    /// way and passes the mechanism to the hook.
257    pub mechanism: u8,
258    /// The endpoint's counter cell.
259    pub cell: &'static PubMonitorCell,
260}
261
262/// phase-503 -- [`MonitorSpec::reaction_id`]: no reaction named.
263pub const REACTION_NONE: u16 = 0;
264/// phase-503 -- [`MonitorSpec::mechanism`]: checked in the executor's spin
265/// loop. Catches a slow route, not a callback that never returns.
266pub const MECHANISM_EXECUTOR: u8 = 0;
267/// phase-503 -- [`MonitorSpec::mechanism`]: an independent timer outside the
268/// executor (the port's), which also catches a stuck callback.
269pub const MECHANISM_TIMER: u8 = 1;
270
271/// phase-503 -- what the reaction hook receives when a supervised route
272/// passes its deadline. `repr(C)`: the C ABI hands the same struct through.
273#[repr(C)]
274#[derive(Debug, Clone, Copy, PartialEq, Eq)]
275pub struct ReactionEvent {
276    /// [`MonitorSpec::reaction_id`] of the row (or, for `hazard-detected`,
277    /// [`HazardSpec::reaction_id`] of the hazard).
278    pub reaction_id: u32,
279    /// Index of the row in the installed monitor table (for
280    /// `hazard-detected`, of the hazard in the installed hazard table).
281    pub row: u32,
282    /// [`rule_code`] of the verdict (`route-deadline-runtime`).
283    pub rule_code: u32,
284    /// Time from the take to the check, ms.
285    pub measured_ms: u32,
286    /// The deadline, ms.
287    pub declared_ms: u32,
288    /// [`MonitorSpec::mechanism`] of the row.
289    pub mechanism: u8,
290    /// play_launch phase 96 W4.2 -- the executor's monotonic µs when the
291    /// verdict fired (for `hazard-detected`, the detection tick: where the
292    /// hazard's rung routes start, so a hook can time its own reaction from
293    /// the same instant). 0 when the executor has no clock.
294    pub fired_at_us: u64,
295}
296
297/// phase-503 -- the reaction hook: called on the spin thread, inside
298/// `spin_once`, once per missed route deadline. It must not block; it starts
299/// the fallback (or hands it to another thread) and returns. `ctx` is what
300/// was registered with the hook.
301pub type ReactionHookFn =
302    unsafe extern "C" fn(event: *const ReactionEvent, ctx: *mut core::ffi::c_void);
303
304/// One monitored subscriber endpoint (W3b.5 age contracts). Separate
305/// table from [`MonitorSpec`] — sub contracts key different endpoints
306/// and need no publish counter.
307#[derive(Debug, Clone, Copy)]
308pub struct AgeMonitorSpec {
309    /// Topic name EXACTLY as the node passes it to `create_subscription`.
310    pub topic: &'static str,
311    /// Endpoint ref for violation reports (the SystemModel contract key).
312    pub fqn: &'static str,
313    /// Declared max take-age (ms). 0 = no age contract.
314    pub max_age_ms: u32,
315    /// phase-462 W5b -- the contract's `deadline` for this endpoint, in the
316    /// wire's milliseconds. 0 = none declared. Judged HERE only when the
317    /// subscription's backend cannot judge it (see
318    /// [`admit_contract_deadline`]); otherwise the backend's own deadline
319    /// event is the detector and this row carries the number unread.
320    pub deadline_ms: u32,
321    /// The endpoint's age accumulator.
322    pub cell: &'static SubMonitorCell,
323}
324
325/// phase-462 W5b -- what a subscription's creation does about its contract
326/// `deadline`, decided from the backend's OWN answer
327/// (`Session::supported_qos_policies`) at the moment it is created.
328///
329/// The served set is a run-time fact -- a C backend answers through the
330/// vtable (issue 1329) -- so it is asked here rather than restated in a build
331/// table that could disagree with it. Three outcomes:
332///
333/// * the backend honours `DEADLINE` and the profile carries the deadline: the
334///   backend judges it, nothing is armed;
335/// * the backend does not honour `DEADLINE`: a profile deadline EQUAL to the
336///   contract's is stripped (it came from the contract, and refusing it at
337///   create would turn a judgeable promise into a boot failure), and the row
338///   is armed so the silence rule judges it;
339/// * the profile never carried the deadline (a registration path that does
340///   not fold the launch overrides): the backend cannot judge what it was
341///   never told, so the row is armed too.
342///
343/// A profile deadline that DIFFERS from the contract's is the caller's own
344/// request and is left alone, so a backend that cannot serve it still refuses
345/// it by name. Returns the profile to create with and the row to arm.
346pub fn admit_contract_deadline(
347    mut qos: nros_rmw::QoSProfile,
348    served: nros_rmw::QoSPolicyMask,
349    ages: &'static [AgeMonitorSpec],
350    topic: &str,
351) -> (nros_rmw::QoSProfile, Option<&'static AgeMonitorSpec>) {
352    let Some(spec) = ages.iter().find(|a| a.topic == topic && a.deadline_ms > 0) else {
353        return (qos, None);
354    };
355    let serves = served.0 & nros_rmw::QoSPolicyMask::DEADLINE.0 != 0;
356    if serves && qos.deadline_ms != 0 {
357        return (qos, None);
358    }
359    if !serves && qos.deadline_ms == spec.deadline_ms {
360        qos.deadline_ms = 0;
361    }
362    (qos, Some(spec))
363}
364
365/// phase-462 W5b -- a released subscription's slot stops being any row's
366/// arrival evidence; its slot may be claimed by an unrelated entry next.
367pub(crate) fn deadline_unbind(ages: &[AgeMonitorSpec], slot: usize) {
368    for spec in ages {
369        let _ = spec.cell.deadline_slot.compare_exchange(
370            slot as u32 + 1,
371            0,
372            Ordering::Relaxed,
373            Ordering::Relaxed,
374        );
375    }
376}
377
378/// phase-462 W5b -- credit one arrival to every armed row whose slot is in
379/// `ready` (the executor's readiness bits for this spin).
380///
381/// Readiness, not dispatch, is the evidence, because it is the one point every
382/// entry passes on the spin thread: an os-priority entry is dispatched on a
383/// worker thread that holds no executor state, and a cooperative one only when
384/// its trigger passes.
385pub(crate) fn credit_arrivals(ages: &[AgeMonitorSpec], ready: u64) {
386    for spec in ages {
387        let slot = spec.cell.deadline_slot.load(Ordering::Relaxed);
388        if slot != 0 && slot <= 64 && ready & (1u64 << (slot - 1)) != 0 {
389            spec.cell.arrivals.fetch_add(1, Ordering::Relaxed);
390        }
391    }
392}
393
394/// phase-436 E4 — how often the stack-headroom rule may query the port, in µs.
395///
396/// On a painted Zephyr stack the query walks every UNUSED byte from the stack
397/// base (`z_stack_space_get`), so querying on every spin put that scan inside
398/// the control loop. The high-water mark only grows, so a throttled check sees
399/// the same low mark and only sees it later. Since E5 the query is also capped
400/// at the declared minimum (`stack_unused_bytes_upto`), so it reads at most
401/// that many bytes.
402pub(crate) const STACK_HEADROOM_CHECK_INTERVAL_US: u64 = 1_000_000;
403
404/// The same limit for a build with no clock: once per this many spins.
405pub(crate) const STACK_HEADROOM_CHECK_SPIN_STRIDE: u32 = 1_000;
406
407/// Whether the stack-headroom query is due.
408///
409/// `last_check_us` is `None` until the first check, which is immediate, so a
410/// bound set at boot is checked on the first spin. After that a clocked build
411/// waits one full interval and a clockless build one full stride of spins.
412pub(crate) fn stack_headroom_check_due(
413    now_us: Option<u64>,
414    last_check_us: Option<u64>,
415    spins_since_check: u32,
416) -> bool {
417    match (last_check_us, now_us) {
418        (None, _) => true,
419        (Some(last), Some(now)) => now.saturating_sub(last) >= STACK_HEADROOM_CHECK_INTERVAL_US,
420        (Some(_), None) => spins_since_check >= STACK_HEADROOM_CHECK_SPIN_STRIDE,
421    }
422}
423
424/// Rate-check window (µs). Matches play_launch's ~5 s time-based trigger
425/// so both runtimes converge on comparable cadence.
426pub const RATE_CHECK_INTERVAL_US: u64 = 5_000_000;
427
428/// Max RATE/LATENCY-monitored endpoints per executor (inline table, no_std).
429///
430/// phase-496 W1 (issue 1471) -- generated, not a literal: the
431/// `NROS_EXECUTOR_MAX_MONITORS` knob, whose derived rung is the image's own
432/// `monitor_rows` count (the sizing descriptor's `[image] monitor_rows`), 8 when nothing
433/// states or derives it. A table longer than this is REFUSED at install
434/// ([`check_table_capacity`]), never truncated: the spin loop inspects only the
435/// first `MAX_MONITORS` specs, so a truncating install would boot with the rest
436/// of the contract silently unwatched.
437pub const MAX_MONITORS: usize = crate::config::MAX_MONITORS;
438/// Max AGE-monitored subscriptions per executor -- the second table, on its
439/// own knob (`NROS_EXECUTOR_MAX_AGE_MONITORS`, derived from `age_rows`) so an
440/// image with rate contracts and no age contracts pays for one table only.
441pub const MAX_AGE_MONITORS: usize = crate::config::MAX_AGE_MONITORS;
442
443/// A monitor table longer than the executor can watch -- phase-496 W1.
444///
445/// Its `Display` NAMES THE KNOB to raise (RFC-0065 D2, refuse and name the
446/// remedy), so every installer that refuses reports the same words.
447#[derive(Debug, Clone, Copy, PartialEq, Eq)]
448pub struct MonitorTableFull {
449    /// `"NROS_EXECUTOR_MAX_MONITORS"` or `"NROS_EXECUTOR_MAX_AGE_MONITORS"`.
450    pub knob: &'static str,
451    /// Rows the table carries.
452    pub rows: usize,
453    /// Rows this build can watch (the knob's resolved value).
454    pub capacity: usize,
455}
456
457impl core::fmt::Display for MonitorTableFull {
458    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
459        write!(
460            f,
461            "monitor table has {} rows but this executor watches {}; raise {} \
462             (it derives from the contract's row count when nothing states it)",
463            self.rows, self.capacity, self.knob
464        )
465    }
466}
467
468/// phase-462 W4 -- a mode's variant table is not the base table's length,
469/// refused by [`super::Executor::set_mode_tables`].
470#[derive(Debug, Clone, Copy, PartialEq, Eq)]
471pub struct ModeTablesMismatch {
472    /// The mode id whose tables differ.
473    pub mode: u16,
474    /// `(the mode's, the base's)` rate/latency row count.
475    pub monitors: (usize, usize),
476    /// `(the mode's, the base's)` age row count.
477    pub ages: (usize, usize),
478    /// phase-462 W4c -- a deadline row naming a SchedContext this executor
479    /// does not have.
480    pub missing_sc: Option<u8>,
481}
482
483impl core::fmt::Display for ModeTablesMismatch {
484    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
485        if let Some(sc) = self.missing_sc {
486            return write!(
487                f,
488                "mode {} moves the deadline of sched context {sc}, which this executor does \
489                 not have: install the mode tables after the sched contexts (re-run the bake)",
490                self.mode
491            );
492        }
493        write!(
494            f,
495            "mode {} bakes {} rate/latency and {} age rows where the base has {} and {}: a \
496             mode is a variant of the base tables, row for row (re-run the bake)",
497            self.mode, self.monitors.0, self.ages.0, self.monitors.1, self.ages.1
498        )
499    }
500}
501
502/// Whether a table of `n_rows` rate/latency rows and `n_ages` age rows fits
503/// this build's executor. The rate table is judged first.
504pub fn check_table_capacity(n_rows: usize, n_ages: usize) -> Result<(), MonitorTableFull> {
505    check_capacity_against(n_rows, n_ages, MAX_MONITORS, MAX_AGE_MONITORS)
506}
507
508/// [`check_table_capacity`] against explicit capacities, so the refusal is
509/// testable at any knob value without rebuilding the crate.
510pub fn check_capacity_against(
511    n_rows: usize,
512    n_ages: usize,
513    max_rows: usize,
514    max_ages: usize,
515) -> Result<(), MonitorTableFull> {
516    if n_rows > max_rows {
517        return Err(MonitorTableFull {
518            knob: "NROS_EXECUTOR_MAX_MONITORS",
519            rows: n_rows,
520            capacity: max_rows,
521        });
522    }
523    if n_ages > max_ages {
524        return Err(MonitorTableFull {
525            knob: "NROS_EXECUTOR_MAX_AGE_MONITORS",
526            rows: n_ages,
527            capacity: max_ages,
528        });
529    }
530    Ok(())
531}
532/// Violation ring depth, per executor, and the slot count of the image-wide
533/// SWD record ([`ViolationRecord`]).
534///
535/// phase-474 I1 -- `NROS_EXECUTOR_MAX_VIOLATIONS` (Zephyr:
536/// `CONFIG_NROS_EXECUTOR_MAX_VIOLATIONS`), default 8. The ring keeps the
537/// LATEST this-many verdicts, so the depth bounds how much recent history a
538/// reader sees and never whether a late violation is kept at all.
539pub const MAX_VIOLATIONS: usize = crate::config::MAX_VIOLATIONS;
540
541/// A detected contract violation, in the play_launch rule-id vocabulary.
542#[derive(Debug, Clone)]
543pub struct Violation {
544    /// `"rate-hierarchy-runtime"` | `"max-age-runtime"` |
545    /// `"max-latency-runtime"` | `"deadline-miss-runtime"` |
546    /// `"stack-headroom-runtime"` | `"alive-supervision-runtime"` |
547    /// `"silence-runtime"` (phase-462 W2) | `"route-latency-runtime"`
548    /// (phase10-B3) | `"route-deadline-runtime"` (phase-503) |
549    /// `"hazard-route-runtime"` (play_launch phase 96 W4.2).
550    pub rule: &'static str,
551    /// Violating endpoint ref (from the spec's `fqn`; the SC name for
552    /// deadline misses).
553    pub fqn: &'static str,
554    /// Measured value. Unit is per-rule: milli-Hz for the rate rule, ms
555    /// for age/latency, µs for deadline misses, dropped activations for
556    /// the timer-overrun rule.
557    pub measured: u32,
558    /// Declared bound, same unit as `measured`.
559    pub declared: u32,
560}
561
562/// phase-474 I1 -- every rule id the executor reports, in WIRE ORDER: a rule's
563/// numeric code is its index here plus one, and 0 means "not a rule this table
564/// knows".
565///
566/// The code is what a trace marker and the SWD record carry, because neither
567/// can carry a string. Append only: a capture or a RAM dump taken before a
568/// reorder would silently decode as the wrong rule, which is the class the
569/// callback-trace marker ids are frozen against (`callback_trace.rs`).
570pub const RULE_IDS: [&str; 16] = [
571    "rate-hierarchy-runtime",
572    "max-age-runtime",
573    "max-latency-runtime",
574    "deadline-miss-runtime",
575    "stack-headroom-runtime",
576    "alive-supervision-runtime",
577    "silence-runtime",
578    "timer-overrun-runtime",
579    "release-jitter-runtime",
580    // phase10-B3.
581    "route-latency-runtime",
582    // phase-503.
583    "route-deadline-runtime",
584    // phase-462 W3b -- the hazard observer, in play_launch's spelling (no
585    // `-runtime` suffix) so one log reader serves both sides of a bridge.
586    "hazard-detected",
587    "hazard-reaction",
588    "hazard-recovered",
589    // phase-462 W4c.
590    "mode-switch-runtime",
591    // play_launch phase 96 W4.2 (phase 85 D4) -- a hazard's rung route.
592    "hazard-route-runtime",
593];
594
595/// The wire code of `rule` ([`RULE_IDS`] index + 1), or 0 for a rule the
596/// table does not know.
597pub fn rule_code(rule: &str) -> u32 {
598    RULE_IDS
599        .iter()
600        .position(|r| *r == rule)
601        .map(|i| i as u32 + 1)
602        .unwrap_or(0)
603}
604
605/// The rule a wire code names, or `None` for 0 / an unknown code.
606pub fn rule_name(code: u32) -> Option<&'static str> {
607    RULE_IDS.get((code as usize).checked_sub(1)?).copied()
608}
609
610/// 32-bit FNV-1a of an endpoint ref -- the identity a trace marker and the SWD
611/// record carry for `Violation::fqn` (a string neither can hold).
612///
613/// FNV-1a because a decoder can recompute it in one line from the model's
614/// endpoint list (`scripts/read-violation-record.py` does), with no table
615/// baked into the image.
616pub const fn fqn_hash(fqn: &str) -> u32 {
617    let b = fqn.as_bytes();
618    let mut h: u32 = 0x811c_9dc5;
619    let mut i = 0;
620    while i < b.len() {
621        h ^= b[i] as u32;
622        h = h.wrapping_mul(0x0100_0193);
623        i += 1;
624    }
625    h
626}
627
628/// phase-474 I1 -- trace marker: a stored violation, `seq << 8 | rule code`.
629/// The id block continues `callback_trace`'s 16-20; ids 1-7 are the
630/// application's and are never used here.
631pub const MARKER_VIOLATION: u32 = 21;
632/// phase-474 I1 -- trace marker: the endpoint ref's [`fqn_hash`].
633pub const MARKER_VIOLATION_FQN: u32 = 22;
634/// phase-474 I1 -- trace marker: `measured`.
635pub const MARKER_VIOLATION_MEASURED: u32 = 23;
636/// phase-474 I1 -- trace marker: `declared`.
637pub const MARKER_VIOLATION_DECLARED: u32 = 24;
638
639/// phase-474 I1 -- the four `(marker_id, arg)` events one stored violation
640/// becomes in a trace (`callback_trace` emits them, in this order, when the
641/// `trace-callbacks` feature is on and a sink is installed). `seq` is the
642/// executor's sequence number; its low 24 bits ride in the first event.
643pub fn violation_marker_words(seq: u32, v: &Violation) -> [(u32, u32); 4] {
644    [
645        (
646            MARKER_VIOLATION,
647            ((seq & 0x00ff_ffff) << 8) | (rule_code(v.rule) & 0xff),
648        ),
649        (MARKER_VIOLATION_FQN, fqn_hash(v.fqn)),
650        (MARKER_VIOLATION_MEASURED, v.measured),
651        (MARKER_VIOLATION_DECLARED, v.declared),
652    ]
653}
654
655/// `"NRVR"` -- nano-ros violation record.
656pub const RECORD_MAGIC: u32 = 0x4e52_5652;
657/// [`ViolationRecord`] layout version. Bump on any field change; a reader
658/// refuses a version it does not know (`scripts/read-violation-record.py`).
659pub const RECORD_VERSION: u32 = 1;
660/// `u32` words in one [`ViolationSlot`].
661pub const RECORD_SLOT_WORDS: u32 = 7;
662/// `u32` words in the [`ViolationRecord`] header, before the first slot.
663pub const RECORD_HEADER_WORDS: u32 = 10;
664
665/// One slot of the SWD record. Every field is a `u32` on every target, so the
666/// layout is the same on a 32-bit board and a 64-bit host.
667#[repr(C)]
668#[derive(Debug)]
669pub struct ViolationSlot {
670    /// 1-based sequence number of the violation in this slot, 0 = empty.
671    /// Written LAST (zeroed first), so a reader that sees a non-zero `seq`
672    /// twice around a read of the other words has a whole entry.
673    pub seq: AtomicU32,
674    /// [`rule_code`].
675    pub rule: AtomicU32,
676    /// [`fqn_hash`] of the endpoint ref.
677    pub fqn_hash: AtomicU32,
678    pub measured: AtomicU32,
679    pub declared: AtomicU32,
680    /// Address of the endpoint ref's bytes (exact on a 32-bit target, the low
681    /// half of it on a 64-bit host) and its length: the text is in the
682    /// image's rodata, so a reader with the ELF can print the name.
683    pub fqn_addr: AtomicU32,
684    pub fqn_len: AtomicU32,
685}
686
687impl ViolationSlot {
688    pub const fn new() -> Self {
689        Self {
690            seq: AtomicU32::new(0),
691            rule: AtomicU32::new(0),
692            fqn_hash: AtomicU32::new(0),
693            measured: AtomicU32::new(0),
694            declared: AtomicU32::new(0),
695            fqn_addr: AtomicU32::new(0),
696            fqn_len: AtomicU32::new(0),
697        }
698    }
699}
700
701impl Default for ViolationSlot {
702    fn default() -> Self {
703        Self::new()
704    }
705}
706
707/// phase-474 I1 -- the image-wide violation record a debugger reads by NAME,
708/// for a board whose console reaches nobody.
709///
710/// The executor's ring is carved out of executor storage (no symbol, and a
711/// drain empties it), so on the Autoware Safety Island it could only be found
712/// by scanning RAM for rule-id string pointers, and it held the FIRST eight
713/// verdicts since boot. This record is the black box beside it: a `#[repr(C)]`
714/// static, every word a `u32`, never drained, keeping the LATEST
715/// [`MAX_VIOLATIONS`] stored violations of every executor in the image, with
716/// the running total.
717///
718/// Layout (version [`RECORD_VERSION`], little-endian `u32` words):
719///
720/// | word | field | meaning |
721/// | --- | --- | --- |
722/// | 0 | `magic` | [`RECORD_MAGIC`] (`"NRVR"`) |
723/// | 1 | `version` | [`RECORD_VERSION`] |
724/// | 2 | `capacity` | slots (`NROS_EXECUTOR_MAX_VIOLATIONS`) |
725/// | 3 | `slot_words` | [`RECORD_SLOT_WORDS`] |
726/// | 4 | `total` | violations stored since boot (the last `seq`) |
727/// | 5 | `head` | slot the NEXT violation takes (`total % capacity`) |
728/// | 6 | `dropped` | entries overwritten (`total - capacity`, saturating) |
729/// | 7 | `suppressed_before_arm` | verdicts before the monitors armed, all executors (phase-474 I2) |
730/// | 8 | `armed` | executors whose monitors armed (counted as each arms; at its first spin by default) (phase-474 I2) |
731/// | 9 | `reserved` | 0 |
732/// | 10.. | `slots` | `capacity` x [`ViolationSlot`] |
733///
734/// The newest entry is in slot `(head + capacity - 1) % capacity`; reading
735/// `capacity` slots backwards from there gives newest to oldest, and a slot
736/// whose `seq` is 0 was never written.
737///
738/// Present only when the image asked for the boot report
739/// (`NROS_BOOT_REPORT=1`, Zephyr `CONFIG_NROS_BOOT_REPORT=y`) -- the same
740/// opt-in as the SWD boot record it sits beside, so an image that does not
741/// opt in is unchanged. Read it with `scripts/read-violation-record.py`.
742#[repr(C)]
743#[derive(Debug)]
744pub struct ViolationRecord<const N: usize> {
745    pub magic: AtomicU32,
746    pub version: AtomicU32,
747    pub capacity: AtomicU32,
748    pub slot_words: AtomicU32,
749    pub total: AtomicU32,
750    pub head: AtomicU32,
751    pub dropped: AtomicU32,
752    pub suppressed_before_arm: AtomicU32,
753    pub armed: AtomicU32,
754    pub reserved: AtomicU32,
755    pub slots: [ViolationSlot; N],
756}
757
758impl<const N: usize> ViolationRecord<N> {
759    /// A valid, empty record: the header is written at compile time, so a
760    /// record with `total == 0` reads as "no violation since boot" rather than
761    /// as uninitialised RAM.
762    pub const fn new() -> Self {
763        Self {
764            magic: AtomicU32::new(RECORD_MAGIC),
765            version: AtomicU32::new(RECORD_VERSION),
766            capacity: AtomicU32::new(N as u32),
767            slot_words: AtomicU32::new(RECORD_SLOT_WORDS),
768            total: AtomicU32::new(0),
769            head: AtomicU32::new(0),
770            dropped: AtomicU32::new(0),
771            suppressed_before_arm: AtomicU32::new(0),
772            armed: AtomicU32::new(0),
773            reserved: AtomicU32::new(0),
774            slots: [const { ViolationSlot::new() }; N],
775        }
776    }
777
778    /// Store one violation, overwriting the oldest when full. Returns its
779    /// 1-based sequence number.
780    ///
781    /// Lock-free and callable from any executor thread: the slot is claimed by
782    /// a `fetch_add` on `total`, so two executors never write one slot unless
783    /// `N` more violations land during one write.
784    pub fn store(&self, v: &Violation) -> u32 {
785        let seq = self.total.fetch_add(1, Ordering::AcqRel).wrapping_add(1);
786        if N == 0 {
787            return seq;
788        }
789        let slot = &self.slots[(seq as usize - 1) % N];
790        slot.seq.store(0, Ordering::Release);
791        slot.rule.store(rule_code(v.rule), Ordering::Relaxed);
792        slot.fqn_hash.store(fqn_hash(v.fqn), Ordering::Relaxed);
793        slot.measured.store(v.measured, Ordering::Relaxed);
794        slot.declared.store(v.declared, Ordering::Relaxed);
795        slot.fqn_addr
796            .store(v.fqn.as_ptr() as usize as u32, Ordering::Relaxed);
797        slot.fqn_len.store(v.fqn.len() as u32, Ordering::Relaxed);
798        slot.seq.store(seq, Ordering::Release);
799        self.head.store(seq % N as u32, Ordering::Relaxed);
800        self.dropped
801            .store(seq.saturating_sub(N as u32), Ordering::Relaxed);
802        seq
803    }
804
805    /// The stored entries, newest first, as `(seq, rule code, fqn hash,
806    /// measured, declared)`. For tests and on-target self-checks; a debugger
807    /// reads the words directly.
808    pub fn newest_first(&self, mut f: impl FnMut(u32, u32, u32, u32, u32)) {
809        let total = self.total.load(Ordering::Acquire);
810        let n = (total as usize).min(N);
811        for k in 0..n {
812            let seq = total - k as u32;
813            let slot = &self.slots[(seq as usize - 1) % N];
814            if slot.seq.load(Ordering::Acquire) != seq {
815                continue; // overwritten or mid-write
816            }
817            f(
818                seq,
819                slot.rule.load(Ordering::Relaxed),
820                slot.fqn_hash.load(Ordering::Relaxed),
821                slot.measured.load(Ordering::Relaxed),
822                slot.declared.load(Ordering::Relaxed),
823            );
824        }
825    }
826}
827
828impl<const N: usize> Default for ViolationRecord<N> {
829    fn default() -> Self {
830        Self::new()
831    }
832}
833
834/// phase-474 I1 -- THE record, by name: `nm zephyr.elf | grep
835/// NROS_VIOLATION_RECORD`. See [`ViolationRecord`] for the layout.
836#[cfg(nros_boot_report)]
837#[unsafe(no_mangle)]
838pub static NROS_VIOLATION_RECORD: ViolationRecord<MAX_VIOLATIONS> = ViolationRecord::new();
839
840/// phase-474 I2 -- application arm requests, image-wide. Each call of
841/// [`request_monitor_arming`] bumps it; an executor waiting for the
842/// application arms when it sees a value other than the one it was opened
843/// with, so a request made before an executor existed does not arm it.
844static ARM_REQUESTS: AtomicU32 = AtomicU32::new(0);
845
846/// phase-474 I2 -- the application says its start-up is over: arm the
847/// contract monitors of every executor in the image that waits for this
848/// (`NROS_MONITOR_ARM_ON_CALL`, or [`Executor::set_monitor_arming`]).
849///
850/// Callable from anywhere -- a node's callback, another thread -- because it
851/// only bumps an atomic; each executor arms at its next spin. On an executor
852/// that is already armed (the default: armed at the first spin) it does
853/// nothing. C: `rclc_monitors_arm()`; C++: `nros::arm_monitors()`.
854///
855/// [`Executor::set_monitor_arming`]: super::Executor::set_monitor_arming
856pub fn request_monitor_arming() {
857    ARM_REQUESTS.fetch_add(1, Ordering::AcqRel);
858}
859
860/// phase-462 W4 -- one contract MODE's variant of an executor's two monitor
861/// tables: the base tables' rows in the base's order, sharing the base's
862/// cells, with the mode's numbers.
863///
864/// Because the rows and cells are the base's, a publisher or subscription
865/// keeps the cell it bound at creation, and a hazard's guard keeps the row
866/// index it names; only the requirement each row is judged against changes.
867/// The bake guarantees that shape (`nros_orchestration_ir::mode_variants`);
868/// [`super::Executor::set_mode_tables`] refuses a table of another length.
869#[derive(Debug, Clone, Copy)]
870pub struct ModeTables {
871    /// The rate/latency table in this mode.
872    pub monitors: &'static [MonitorSpec],
873    /// The age table in this mode.
874    pub ages: &'static [AgeMonitorSpec],
875    /// phase-462 W4c -- the SchedContext deadlines this mode moves. Every
876    /// mode's table, the base's included, lists the SAME contexts, so entering
877    /// any mode leaves every moved deadline at that mode's value.
878    pub deadlines: &'static [ModeDeadline],
879    /// phase-462 W4c -- how long ENTERING this mode may take, ms: its reaction
880    /// path's `max_latency`. 0 = the contract states none.
881    pub switch_budget_ms: u32,
882    /// phase-462 W4c -- that reaction path's key, the `mode-switch-runtime`
883    /// report's endpoint ref ("" when there is none).
884    pub reaction: &'static str,
885}
886
887/// phase-462 W4c -- one SchedContext's deadline in one mode.
888///
889/// The deadline is the one tier fact a switch can apply: the executor reads
890/// `SchedContext::deadline_us` each time it judges a dispatch. A priority is a
891/// thread's and a reservation is armed at boot, so the bake lists a mode that
892/// moves either instead of baking it.
893///
894/// `repr(C)`: the C ABI's `nros_cpp_mode_deadline_t` is the same two fields.
895#[repr(C)]
896#[derive(Debug, Clone, Copy, PartialEq, Eq)]
897pub struct ModeDeadline {
898    /// The SchedContext's id on this executor.
899    pub sc: u8,
900    /// The deadline in this mode, µs; 0 = none.
901    pub deadline_us: u32,
902}
903
904/// phase-462 W4c -- the `mode-switch-runtime` verdict for entering `t`: how
905/// long the request could have waited, against the mode's budget.
906///
907/// An executor enters a requested mode at its next spin, so a request
908/// waited at most the gap since the previous spin's check -- the measured
909/// value is that gap, an upper bound on the switch's own latency. `None` when
910/// the mode states no budget, the gap is within it, or there is no clock.
911pub(crate) fn check_mode_switch(t: &ModeTables, waited_ms: Option<u32>) -> Option<Violation> {
912    let waited = waited_ms?;
913    (t.switch_budget_ms != 0 && waited > t.switch_budget_ms).then_some(Violation {
914        rule: "mode-switch-runtime",
915        fqn: t.reaction,
916        measured: waited,
917        declared: t.switch_budget_ms,
918    })
919}
920
921/// phase-462 W4 -- the mode the application asked for. A STATE, not an
922/// event: an executor opened after the request enters it at its first spin.
923static MODE_REQUEST: AtomicU32 = AtomicU32::new(0);
924
925/// phase-462 W4 -- ask every executor to judge by mode `id`'s tables.
926///
927/// `id` is the baked mode id (`0` = the base contract, `1..` the contract's
928/// modes in key order; the entry names each as a constant). Like
929/// [`request_monitor_arming`] this only stores an atomic: each executor
930/// swaps its own tables at its next spin, so the switch is bounded by the
931/// slowest executor's spin period and never races a rule mid-judgement. An
932/// id an executor has no table for leaves it on the tables it has, and it
933/// says so once.
934pub fn request_mode(id: u16) {
935    MODE_REQUEST.store(u32::from(id), Ordering::Release);
936}
937
938/// The mode last asked for with [`request_mode`] (0 when never asked).
939#[must_use]
940pub fn requested_mode() -> u16 {
941    MODE_REQUEST.load(Ordering::Acquire) as u16
942}
943
944/// The current arm-request count (for an executor to remember at open).
945pub(crate) fn arm_requests() -> u32 {
946    ARM_REQUESTS.load(Ordering::Acquire)
947}
948
949/// phase-474 I2 -- when an executor's contract monitors start judging.
950///
951/// A verdict reached before arming is COUNTED
952/// ([`Executor::violations_suppressed_before_arm`], and the SWD record's
953/// `suppressed_before_arm`) but not stored, logged, sunk or traced: start-up
954/// -- registration, the first join, inputs that have not started yet -- looks
955/// like overruns, jitter and silence, and judging it as steady state filled
956/// the island's ring with eight start-up entries before any act ran.
957///
958/// The default is read from Kconfig / the build environment only
959/// (`NROS_MONITOR_ARM_ON_CALL`, `NROS_MONITOR_ARM_GRACE_MS`); the contract
960/// carries no key for it.
961///
962/// [`Executor::violations_suppressed_before_arm`]: super::Executor::violations_suppressed_before_arm
963#[derive(Debug, Clone, Copy, PartialEq, Eq)]
964pub struct MonitorArming {
965    /// Wait for [`request_monitor_arming`] / `Executor::arm_monitors`.
966    /// `false` (the default) arms at the first spin, today's behaviour.
967    pub on_call: bool,
968    /// Arm anyway this many ms after the first spin; 0 = no deadline. With
969    /// `on_call` it is the backstop for an application that never calls; on
970    /// its own it is a grace period for an application with no lifecycle.
971    pub grace_ms: u32,
972}
973
974impl MonitorArming {
975    /// The build's default (`NROS_MONITOR_ARM_ON_CALL`,
976    /// `NROS_MONITOR_ARM_GRACE_MS`).
977    pub const DEFAULT: Self = Self {
978        on_call: crate::config::MONITOR_ARM_ON_CALL,
979        grace_ms: crate::config::MONITOR_ARM_GRACE_MS,
980    };
981
982    /// Armed from the start: no wait for a call and no grace period.
983    pub const fn armed_at_first_spin(self) -> bool {
984        !self.on_call && self.grace_ms == 0
985    }
986}
987
988/// Issue 1635 -- where an executor hands its violations when the image asked
989/// it to (`Executor::set_violation_sink`).
990///
991/// A function and an opaque context rather than a closure, because the
992/// executor is not generic and the reporter it reaches lives in a crate above
993/// this one (`nros-cpp` publishes on `/diagnostics` through
994/// `nros-diagnostics`). Called from inside the executor's spin (never from
995/// inside a user callback) -- so the sink may publish.
996///
997/// phase-474 I7 -- called OFF the judged dispatch, not at detection: a stored
998/// verdict is queued ([`REPORT_QUEUE`] deep) and the queue is handed over at
999/// the end of a spin whose next timer is further away than the last report
1000/// cost (or once a verdict has waited [`REPORT_DEFER_MAX_US`]). The sink gets
1001/// every pending verdict, oldest first, and returns how many of them it took
1002/// -- a PREFIX, so it may coalesce as many as fit one message and leave the
1003/// rest for the next slot. Returning 0 is a failure the sink has counted;
1004/// the executor then drops the oldest pending verdict so a report that can
1005/// never be sent cannot stall the queue (it stays in the ring and the record).
1006///
1007/// On the safety island the report at detection cost 15-20 ms of spin per
1008/// verdict over a 921,600-baud serial link, inside the tick being judged, and
1009/// that cost tripped `release-jitter-runtime` and `timer-overrun-runtime`:
1010/// one verdict bred more.
1011///
1012/// # Safety
1013/// The sink is called with the `ctx` it was installed with; whoever installs it
1014/// guarantees `ctx` is valid for every call until the sink is replaced or the
1015/// executor is dropped.
1016pub type ViolationSink = unsafe fn(ctx: *mut core::ffi::c_void, pending: &[Violation]) -> usize;
1017
1018/// phase-474 I7 -- how many stored verdicts wait for the sink. A verdict
1019/// stored while the queue is full evicts the oldest pending one, which is
1020/// counted (`ReportQueue::overflowed`); both stay in the ring and the SWD
1021/// record. Four is the `DiagnosticArray` status capacity of the generated
1022/// type, so one coalesced report never needs more.
1023pub const REPORT_QUEUE: usize = 4;
1024
1025/// phase-474 I7 -- the longest a pending verdict waits for a slot with slack
1026/// before it is reported anyway, in microseconds. Bounds the deferral on an
1027/// executor whose timers never leave a gap as long as the report costs.
1028pub const REPORT_DEFER_MAX_US: u64 = 1_000_000;
1029
1030/// phase-474 I7 -- verdicts stored but not yet handed to the sink, and what
1031/// the hand-over has cost. One struct so the executor size test can name it.
1032#[derive(Debug, Default)]
1033pub(crate) struct ReportQueue {
1034    /// Oldest first.
1035    pub(crate) pending: heapless::Vec<Violation, REPORT_QUEUE>,
1036    /// Monotonic us at which the oldest pending verdict was first seen by a
1037    /// flush, for [`REPORT_DEFER_MAX_US`].
1038    pub(crate) since_us: Option<u64>,
1039    /// What the last hand-over took, and the worst one, in us.
1040    pub(crate) last_cost_us: u32,
1041    pub(crate) max_cost_us: u32,
1042    /// Pending verdicts evicted by a newer one before any slot came (saturating).
1043    pub(crate) overflowed: u32,
1044    /// Verdicts the sink was handed and could not send (saturating).
1045    pub(crate) failed: u32,
1046    /// The reporter's own spin time since the last release-jitter sample: it
1047    /// is the monitor's cost, stated by `max_cost_us`, not lateness of the
1048    /// application's release, so the jitter rule does not charge it.
1049    pub(crate) debt_us: u64,
1050}
1051
1052/// phase-474 I1 -- everything an executor does with a detected violation, in
1053/// one place: the switches, the sink, the ring and its counters.
1054///
1055/// The ring keeps the LATEST [`MAX_VIOLATIONS`] verdicts: a push into a full
1056/// ring evicts the oldest and counts it in `dropped`. It used to refuse the
1057/// push instead, so a board whose start-up filled the ring stored nothing
1058/// after it (the Autoware Safety Island's W31 bring-up: 8 of 8 slots, all
1059/// start-up). `total` counts every verdict stored since boot, so
1060/// `total - dropped - ring.len()` is how many a drain has already taken.
1061pub(crate) struct ViolationChannel<'s> {
1062    /// Issue 0514 -- log each violation at detection.
1063    pub(crate) report: bool,
1064    /// phase-474 I1 -- drain the ring at the end of every spin and report each
1065    /// entry with its sequence number and the counters
1066    /// ([`Executor::set_violation_drain_report`]). Replaces the log at
1067    /// detection while on, so one verdict is one line.
1068    ///
1069    /// [`Executor::set_violation_drain_report`]: super::Executor::set_violation_drain_report
1070    pub(crate) drain_report: bool,
1071    /// Issue 1635 -- the image's reporter. The context is a `usize` so the
1072    /// executor keeps its auto traits; it is the installer's pointer.
1073    /// phase-474 I7 -- fed from `reports`, off the judged dispatch.
1074    pub(crate) sink: Option<(ViolationSink, usize)>,
1075    /// phase-474 I7 -- what waits for the sink.
1076    pub(crate) reports: ReportQueue,
1077    /// phase-409 -- CARVED, at `MAX_VIOLATIONS` (no `ExecutorSizing` knob:
1078    /// the build-time depth is the capability).
1079    pub(crate) ring: super::storage::CarvedVec<'s, Violation>,
1080    /// phase-474 I1/I2 -- the counters and the arming state.
1081    pub(crate) counts: ViolationCounts,
1082}
1083
1084/// phase-474 I1/I2 -- the scalars of a [`ViolationChannel`]: its counters and
1085/// its arming state. One struct so `the_executor_value_does_not_scale_with_the_knobs`
1086/// can name what they cost; none of them scales with a knob.
1087#[derive(Debug, Clone, Copy)]
1088pub(crate) struct ViolationCounts {
1089    /// Violations this executor stored since boot (wrapping). It is also the
1090    /// sequence number of the newest entry in `ring`: every stored verdict is
1091    /// pushed, so the ring holds `total - len + 1 ..= total`.
1092    pub(crate) total: u32,
1093    /// Issue 0514 / phase-474 I1 -- entries evicted from the full ring before
1094    /// any drain took them (saturating).
1095    pub(crate) dropped: u32,
1096    /// phase-474 I2 -- whether verdicts are stored. Until then they are only
1097    /// counted in `suppressed_before_arm`.
1098    pub(crate) armed: bool,
1099    /// phase-474 I2 -- verdicts reached before arming (saturating).
1100    pub(crate) suppressed_before_arm: u32,
1101    /// phase-474 I2 -- the arming policy ([`MonitorArming`]), kept as its two
1102    /// fields so the flags pack beside the ones above.
1103    pub(crate) arm_on_call: bool,
1104    pub(crate) arm_grace_ms: u32,
1105    /// phase-474 I2 -- whether the executor has spun, and the monotonic ms
1106    /// (wrapping `u32`) of its first spin, which the grace deadline counts
1107    /// from.
1108    pub(crate) spun: bool,
1109    pub(crate) first_spin_ms: u32,
1110    /// phase-474 I2 -- [`arm_requests`] when the executor opened; a different
1111    /// value later is the application's call.
1112    pub(crate) arm_requests_seen: u32,
1113}
1114
1115impl<'s> ViolationChannel<'s> {
1116    pub(crate) fn new(ring: super::storage::CarvedVec<'s, Violation>) -> Self {
1117        Self {
1118            report: true,
1119            drain_report: crate::config::VIOLATION_DRAIN_REPORT,
1120            sink: None,
1121            reports: ReportQueue::default(),
1122            ring,
1123            counts: ViolationCounts {
1124                total: 0,
1125                dropped: 0,
1126                armed: MonitorArming::DEFAULT.armed_at_first_spin(),
1127                suppressed_before_arm: 0,
1128                arm_on_call: MonitorArming::DEFAULT.on_call,
1129                arm_grace_ms: MonitorArming::DEFAULT.grace_ms,
1130                spun: false,
1131                first_spin_ms: 0,
1132                arm_requests_seen: arm_requests(),
1133            },
1134        }
1135    }
1136
1137    /// Issue 1635 / phase-474 I1 -- THE one place a detected violation goes:
1138    /// the log floor (issue 0514) unless the drain hook reports instead, the
1139    /// image's sink's queue when one is installed (sent later, phase-474 I7),
1140    /// the trace marker when callback
1141    /// tracing is compiled in, the SWD record when the image keeps one, and
1142    /// the ring [`drain_violations`] reads.
1143    ///
1144    /// [`drain_violations`]: super::Executor::drain_violations
1145    pub(crate) fn record(&mut self, v: Violation) {
1146        if !self.counts.armed {
1147            // phase-474 I2 -- start-up: counted, never stored.
1148            self.counts.suppressed_before_arm = self.counts.suppressed_before_arm.saturating_add(1);
1149            #[cfg(nros_boot_report)]
1150            NROS_VIOLATION_RECORD
1151                .suppressed_before_arm
1152                .fetch_add(1, Ordering::Relaxed);
1153            return;
1154        }
1155        if self.report && !self.drain_report {
1156            log_violation(&v);
1157        }
1158        if self.sink.is_some() {
1159            // phase-474 I7 -- queued, not sent: the report leaves at the end of
1160            // a spin with slack (`Executor::flush_violation_reports`), so it
1161            // never costs the dispatch being judged.
1162            if self.reports.pending.is_full() {
1163                self.reports.pending.remove(0);
1164                self.reports.overflowed = self.reports.overflowed.saturating_add(1);
1165            }
1166            let _ = self.reports.pending.push(v.clone());
1167        }
1168        // `total` is this executor's sequence number for the verdict: the log
1169        // line and the trace marker carry it, so the two name one verdict by
1170        // one number. The SWD record numbers across every executor in the
1171        // image; on a single-executor image the two coincide.
1172        self.counts.total = self.counts.total.wrapping_add(1);
1173        #[cfg(nros_boot_report)]
1174        NROS_VIOLATION_RECORD.store(&v);
1175        #[cfg(feature = "trace-callbacks")]
1176        super::callback_trace::violation(self.counts.total, &v);
1177        if self.ring.capacity() == 0 {
1178            self.counts.dropped = self.counts.dropped.saturating_add(1);
1179            return;
1180        }
1181        if self.ring.len() == self.ring.capacity() {
1182            self.ring.pop_front();
1183            self.counts.dropped = self.counts.dropped.saturating_add(1);
1184        }
1185        // Cannot fail: a slot was just made if the ring was full.
1186        let _ = self.ring.push(v);
1187    }
1188
1189    /// Hand every entry still in the ring to `f`, oldest first, with its
1190    /// sequence number, and empty the ring.
1191    pub(crate) fn drain(&mut self, mut f: impl FnMut(u32, &Violation)) {
1192        let n = self.ring.len() as u32;
1193        let first = self.counts.total.wrapping_sub(n).wrapping_add(1);
1194        for (k, v) in self.ring.iter().enumerate() {
1195            f(first.wrapping_add(k as u32), v);
1196        }
1197        self.ring.clear();
1198    }
1199
1200    /// phase-474 I1 -- the drain-and-report hook: one log line per entry with
1201    /// its sequence number and the counters, so a reader of the log knows how
1202    /// many it missed. Called at the end of a spin when `drain_report` is on.
1203    pub(crate) fn drain_and_report(&mut self) {
1204        if !self.drain_report || self.ring.is_empty() {
1205            return;
1206        }
1207        let (total, dropped) = (self.counts.total, self.counts.dropped);
1208        self.drain(|seq, v| log_drained_violation(seq, total, dropped, v));
1209    }
1210}
1211
1212/// Per-spec accounting state (parallel to the spec table).
1213#[derive(Debug, Clone, Copy, Default)]
1214pub(crate) struct MonitorState {
1215    /// Window opened (a plain bool, not a 0-sentinel on the timestamp —
1216    /// `now_us == 0` is a legitimate first sample on freshly-started
1217    /// monotonic clocks).
1218    pub(crate) opened: bool,
1219    pub(crate) window_start_us: u64,
1220    pub(crate) count_at_window_start: u32,
1221    /// Suppress duplicate reports: only re-report after a clean window.
1222    pub(crate) violated_last_window: bool,
1223    /// W3b.5 — separate dedup for the latency rule on the same spec row.
1224    pub(crate) latency_violated_last_window: bool,
1225    /// phase-474 I8 -- the open window has been moved onto the stream's
1226    /// phase (or judged as opened). See [`check_rate`].
1227    pub(crate) aligned: bool,
1228}
1229
1230/// Pure rate check over one window boundary. Returns `Some(violation)`
1231/// when the window elapsed AND the measured rate is below the declared
1232/// minimum (and we didn't already report last window).
1233///
1234/// Extracted from the executor so the math is unit-testable without a
1235/// session: publish counting is injected via the cell, time via `now_us`.
1236pub(crate) fn check_rate(
1237    spec: &MonitorSpec,
1238    state: &mut MonitorState,
1239    now_us: u64,
1240) -> Option<Violation> {
1241    if spec.min_rate_hz_milli == 0 {
1242        return None;
1243    }
1244    let count = spec.cell.count.load(Ordering::Relaxed);
1245    if !state.opened {
1246        // First observation: open the window, no verdict yet.
1247        state.opened = true;
1248        state.aligned = false;
1249        state.window_start_us = now_us;
1250        state.count_at_window_start = count;
1251        return None;
1252    }
1253    let window_us = now_us.saturating_sub(state.window_start_us);
1254    if !state.aligned && window_us < RATE_CHECK_INTERVAL_US {
1255        // phase-474 I8 -- the first window starts at the first tick that sees
1256        // a new sample, not where it was opened. The tick runs before the
1257        // spin's dispatch, so in steady state every boundary sits one tick
1258        // after a publish and the count matches the span. The window opened
1259        // at boot or by arming sits wherever that spin fell: off the
1260        // stream's phase it spans N samples over N periods plus the offset,
1261        // and a stream at exactly its declared rate read short (the island:
1262        // 9984 of 10000 mHz, once, after arming). Re-anchoring here puts the
1263        // first window on the same phase the later ones have. A stream that
1264        // stays silent is not held open: once a whole window passes with no
1265        // sample it is judged as opened (below), at 0.
1266        if count != state.count_at_window_start {
1267            state.aligned = true;
1268            state.window_start_us = now_us;
1269            state.count_at_window_start = count;
1270        }
1271        return None;
1272    }
1273    state.aligned = true;
1274    if window_us < RATE_CHECK_INTERVAL_US {
1275        return None;
1276    }
1277    let published = count.wrapping_sub(state.count_at_window_start) as u64;
1278    // milli-Hz = published * 1e3 / window_s = published * 1e9 / window_us
1279    let measured_milli_hz =
1280        (published.saturating_mul(1_000_000_000) / window_us.max(1)).min(u32::MAX as u64) as u32;
1281    // phase-474 I9 -- judge the count, not the floored quotient. The window
1282    // closes at the first tick that finds it elapsed, so `window_us` is the
1283    // declared interval plus that tick's lateness, while a stream at exactly
1284    // its declared minimum has published one sample per period: 50 over
1285    // 5.0005 s reads 9999 of 10000 mHz and was judged slow (the island, every
1286    // window, reported once thanks to `violated_last_window`). A stream at
1287    // the declared minimum delivers at least floor(min_rate * window) samples
1288    // in any window, whatever its phase; fewer than that is slow.
1289    let required = (spec.min_rate_hz_milli as u64).saturating_mul(window_us) / 1_000_000_000;
1290
1291    // Roll the window.
1292    state.window_start_us = now_us;
1293    state.count_at_window_start = count;
1294
1295    if published < required {
1296        if state.violated_last_window {
1297            return None; // still violated — already reported
1298        }
1299        state.violated_last_window = true;
1300        Some(Violation {
1301            rule: "rate-hierarchy-runtime",
1302            fqn: spec.fqn,
1303            measured: measured_milli_hz,
1304            declared: spec.min_rate_hz_milli,
1305        })
1306    } else {
1307        state.violated_last_window = false;
1308        None
1309    }
1310}
1311
1312/// Pure latency check: drains the spec cell's window-max take→publish
1313/// latency and fires when it exceeds the declared node-path budget.
1314/// Same report-once-until-recovery semantics as the rate rule; runs on
1315/// every monitor tick (the cell accumulates between ticks, so no window
1316/// bookkeeping is needed — draining IS the window roll).
1317pub(crate) fn check_latency(spec: &MonitorSpec, state: &mut MonitorState) -> Option<Violation> {
1318    if spec.max_latency_ms == 0 {
1319        return None;
1320    }
1321    let max_us = spec.cell.max_latency_us.swap(0, Ordering::Relaxed);
1322    let max_ms = max_us / 1_000;
1323    if max_ms > spec.max_latency_ms {
1324        if state.latency_violated_last_window {
1325            return None;
1326        }
1327        state.latency_violated_last_window = true;
1328        Some(Violation {
1329            rule: "max-latency-runtime",
1330            fqn: spec.fqn,
1331            measured: max_ms,
1332            declared: spec.max_latency_ms,
1333        })
1334    } else {
1335        // A quiet window (no dispatch attributed) also counts as clean —
1336        // recovery resets the dedup like the rate rule's clean window.
1337        state.latency_violated_last_window = false;
1338        None
1339    }
1340}
1341
1342/// phase10-B3 -- bind a subscription slot as the route trigger of every row
1343/// whose [`MonitorSpec::route_trigger`] is `topic`. Called once per
1344/// registration; the first subscription on the topic keeps the binding.
1345pub(crate) fn route_bind(table: &[MonitorSpec], topic: &str, slot: usize) {
1346    if slot >= ROUTE_SLOT_MASK as usize {
1347        return;
1348    }
1349    for spec in table.iter().take(MAX_MONITORS) {
1350        if spec.route_max_latency_ms == 0 || spec.route_trigger != topic {
1351            continue;
1352        }
1353        let w = spec.cell.route.load(Ordering::Relaxed);
1354        if w & ROUTE_SLOT_MASK == 0 {
1355            spec.cell.route.store(
1356                (w & !ROUTE_SLOT_MASK) | (slot as u32 + 1),
1357                Ordering::Relaxed,
1358            );
1359        }
1360    }
1361}
1362
1363/// phase10-B3 -- forget a released slot's route bindings and any take it left
1364/// waiting.
1365pub(crate) fn route_unbind(table: &[MonitorSpec], slot: usize) {
1366    for spec in table.iter().take(MAX_MONITORS) {
1367        let w = spec.cell.route.load(Ordering::Relaxed);
1368        if w & ROUTE_SLOT_MASK == slot as u32 + 1 {
1369            spec.cell.route.store(0, Ordering::Relaxed);
1370        }
1371    }
1372}
1373
1374/// phase10-B3 -- the dispatch in `slot` took a sample at `take_us`: every
1375/// route whose trigger is that slot starts its clock, unless an earlier take
1376/// is still waiting for the output (the route is judged from the EARLIEST
1377/// unanswered take: a take that arrives just after a tick waits for the whole
1378/// next period, and that is the wait the budget has to hold).
1379pub(crate) fn route_take(table: &[MonitorSpec], slot: usize, take_us: u64) {
1380    let bound = slot as u32 + 1;
1381    for spec in table.iter().take(MAX_MONITORS) {
1382        let w = spec.cell.route.load(Ordering::Relaxed);
1383        if w & ROUTE_SLOT_MASK != bound || w & ROUTE_PENDING != 0 {
1384            continue;
1385        }
1386        spec.cell
1387            .route_take_us
1388            .store(take_us as u32, Ordering::Relaxed);
1389        // Release: a `RouteWatch` on another thread that sees the route
1390        // pending sees this take with it.
1391        spec.cell.route.store(w | ROUTE_PENDING, Ordering::Release);
1392    }
1393}
1394
1395/// phase10-B3 -- a dispatch ended at `end_us`. Every route output whose
1396/// counter advanced during it (`counts_before`, as the latency rule reads it)
1397/// and that has a take waiting is judged: take to dispatch end, an upper bound
1398/// on take to that publish, against the path's budget. Over budget is
1399/// reported once until a route is within it again.
1400///
1401/// phase-503 -- a SUPERVISED route ([`MonitorSpec::route_deadline_ms`])
1402/// answered at or after its deadline is also overdue: it is reported as
1403/// `route-deadline-runtime` and its reaction requested, as `route_overdue`
1404/// would have done had it run in time. Under `MECHANISM_EXECUTOR` this is the
1405/// only detection a single long callback gets: the callback that holds the
1406/// spin thread past the deadline publishes the output before the executor's
1407/// overdue check can run, so the miss is caught here, at completion -- late
1408/// by the callback's overrun. A route whose deadline was already caught
1409/// (`route_overdue` dropped its take) is not pending any more and is not
1410/// judged again, so each take gets at most one deadline verdict.
1411///
1412/// `report(row, violation, record, react)`: `record` is `false` while the
1413/// route is still marked violated from an earlier verdict (report once until
1414/// a route is within its budget again); `react` asks the caller to call the
1415/// reaction hook, on every late supervised take as `route_overdue` does.
1416pub(crate) fn route_judge(
1417    table: &[MonitorSpec],
1418    counts_before: &[u32; MAX_MONITORS],
1419    end_us: u64,
1420    mut report: impl FnMut(usize, Violation, bool, bool),
1421) {
1422    for (k, spec) in table.iter().take(MAX_MONITORS).enumerate() {
1423        if spec.route_max_latency_ms == 0 {
1424            continue;
1425        }
1426        let w = spec.cell.route.load(Ordering::Relaxed);
1427        if w & ROUTE_PENDING == 0 || spec.cell.count.load(Ordering::Relaxed) == counts_before[k] {
1428            continue;
1429        }
1430        let took_us = (end_us as u32).wrapping_sub(spec.cell.route_take_us.load(Ordering::Relaxed));
1431        let measured_ms = took_us / 1_000;
1432        let first = w & ROUTE_VIOLATED == 0;
1433        let late = measured_ms > spec.route_max_latency_ms;
1434        let overdue =
1435            route_supervised(spec) && took_us >= spec.route_deadline_ms.saturating_mul(1_000);
1436        let mut next = w & !ROUTE_PENDING;
1437        if late && first {
1438            report(
1439                k,
1440                Violation {
1441                    rule: "route-latency-runtime",
1442                    fqn: spec.fqn,
1443                    measured: measured_ms,
1444                    declared: spec.route_max_latency_ms,
1445                },
1446                true,
1447                false,
1448            );
1449        }
1450        if overdue {
1451            report(
1452                k,
1453                Violation {
1454                    rule: "route-deadline-runtime",
1455                    fqn: spec.fqn,
1456                    measured: measured_ms,
1457                    declared: spec.route_deadline_ms,
1458                },
1459                first,
1460                true,
1461            );
1462        }
1463        if late || overdue {
1464            next |= ROUTE_VIOLATED;
1465        } else {
1466            next &= !ROUTE_VIOLATED;
1467        }
1468        spec.cell.route.store(next, Ordering::Relaxed);
1469    }
1470}
1471
1472/// phase-503 -- whether a pending route of `spec` has a supervised deadline.
1473#[inline]
1474fn route_supervised(spec: &MonitorSpec) -> bool {
1475    spec.route_deadline_ms > 0 && spec.route_max_latency_ms > 0
1476}
1477
1478/// phase-503 -- time from now to the earliest deadline of a pending supervised
1479/// route, in microseconds (0 when one is already due), or `None` when no
1480/// supervised route is waiting. This is the executor's wake source for the
1481/// rule, and what a port's independent timer (`MECHANISM_TIMER`) arms on.
1482///
1483/// The due time is `take + route_deadline_ms`, derived from the stored take
1484/// rather than kept in a second cell: both are the clock's low 32 bits, so the
1485/// difference is exact across a wrap as long as a route waits less than ~71
1486/// minutes, and a supervised one is reported long before that.
1487pub fn route_next_due_us(table: &[MonitorSpec], now_us: u64) -> Option<u64> {
1488    route_next_due(table, || now_us)
1489}
1490
1491/// [`route_next_due_us`] with the clock read AFTER each pending row's take is
1492/// loaded, so a take stored by another thread between the two reads can never
1493/// look older than now (which would read as a due already passed).
1494fn route_next_due(table: &[MonitorSpec], now_us: impl Fn() -> u64) -> Option<u64> {
1495    let mut best: Option<u64> = None;
1496    for spec in table.iter().take(MAX_MONITORS) {
1497        if !route_supervised(spec) {
1498            continue;
1499        }
1500        let w = spec.cell.route.load(Ordering::Acquire);
1501        if w & ROUTE_PENDING == 0 {
1502            continue;
1503        }
1504        let take_us = spec.cell.route_take_us.load(Ordering::Relaxed);
1505        let waited_us = (now_us() as u32).wrapping_sub(take_us);
1506        let deadline_us = spec.route_deadline_ms.saturating_mul(1_000);
1507        let left = deadline_us.saturating_sub(waited_us) as u64;
1508        best = Some(best.map_or(left, |b| b.min(left)));
1509        if left == 0 {
1510            break;
1511        }
1512    }
1513    best
1514}
1515
1516/// phase-503 W7 -- an executor's supervised routes, read from OUTSIDE it: from
1517/// a port's timer expiry, a reaction thread or an interrupt, while the spin
1518/// thread is anywhere, including inside a dispatch.
1519///
1520/// A route is answered when the dispatch that publishes its output ends:
1521/// `route_judge` clears [`ROUTE_PENDING`] right there, on the executor's
1522/// clock, in the row's atomic cell. Until this type, a port's independent timer
1523/// (`MECHANISM_TIMER`) learned that only from `Executor::route_next_due_us`
1524/// after the spin that answered it RETURNED, so a callback that answered
1525/// before the due inside a spin that ran past it looked like a miss at the
1526/// timer's expiry (a false safe state on a board). [`Self::next_due_us`] reads
1527/// the cells at the moment of the call instead: `Some(0)` only when a take is
1528/// still unanswered at or after its deadline -- the executor's own verdict
1529/// (`route_overdue`, `route_judge`: an answer exactly at the deadline is a
1530/// miss).
1531///
1532/// It never touches the executor: the table and the clock it holds are fixed
1533/// once the monitors are installed (take it after `set_monitor_table` and any
1534/// clock change). Reading takes no lock and allocates nothing.
1535#[derive(Debug, Clone, Copy)]
1536pub struct RouteWatch {
1537    table: &'static [MonitorSpec],
1538    clock_us: fn() -> u64,
1539}
1540
1541impl RouteWatch {
1542    /// A watch over `table` on `clock_us` (microseconds, the clock the
1543    /// executor stamps takes with).
1544    pub const fn new(table: &'static [MonitorSpec], clock_us: fn() -> u64) -> Self {
1545        Self { table, clock_us }
1546    }
1547
1548    /// The watched table.
1549    pub const fn table(&self) -> &'static [MonitorSpec] {
1550        self.table
1551    }
1552
1553    /// The clock the watch reads.
1554    pub const fn clock_us(&self) -> fn() -> u64 {
1555        self.clock_us
1556    }
1557
1558    /// Microseconds from now to the earliest deadline of a pending supervised
1559    /// route; `Some(0)` when one has passed with no answer; `None` when none
1560    /// is waiting (every take answered, or none taken).
1561    pub fn next_due_us(&self) -> Option<u64> {
1562        route_next_due(self.table, self.clock_us)
1563    }
1564}
1565
1566/// phase-503 -- judge every pending supervised route at `now_us`: a take still
1567/// unanswered at or after `take + route_deadline_ms` is OVERDUE. Its waiting
1568/// take is dropped (the route stops pending, so the next take starts a new
1569/// one) and the route is marked [`ROUTE_VIOLATED`], as `route_judge` marks an
1570/// over-budget route.
1571///
1572/// `react(row, violation, first)` is called for every overdue route: `first`
1573/// is `false` while the route is still marked violated from an earlier verdict
1574/// (report once until a route is within its budget again, the latency rule's
1575/// rule), and the caller records the violation only when it is `true`. The
1576/// reaction is requested every time, since each overdue take is a missed
1577/// deadline the fallback has to cover.
1578pub(crate) fn route_overdue(
1579    table: &[MonitorSpec],
1580    now_us: u64,
1581    mut react: impl FnMut(usize, Violation, bool),
1582) {
1583    for (k, spec) in table.iter().take(MAX_MONITORS).enumerate() {
1584        if !route_supervised(spec) {
1585            continue;
1586        }
1587        let w = spec.cell.route.load(Ordering::Relaxed);
1588        if w & ROUTE_PENDING == 0 {
1589            continue;
1590        }
1591        let waited_us =
1592            (now_us as u32).wrapping_sub(spec.cell.route_take_us.load(Ordering::Relaxed));
1593        if waited_us < spec.route_deadline_ms.saturating_mul(1_000) {
1594            continue;
1595        }
1596        spec.cell
1597            .route
1598            .store((w & !ROUTE_PENDING) | ROUTE_VIOLATED, Ordering::Relaxed);
1599        react(
1600            k,
1601            Violation {
1602                rule: "route-deadline-runtime",
1603                fqn: spec.fqn,
1604                measured: waited_us / 1_000,
1605                declared: spec.route_deadline_ms,
1606            },
1607            w & ROUTE_VIOLATED == 0,
1608        );
1609    }
1610}
1611
1612/// phase-503 -- whether any row carries a supervised route deadline.
1613pub(crate) fn route_deadline_active(table: &[MonitorSpec]) -> bool {
1614    table.iter().any(route_supervised)
1615}
1616
1617/// phase10-B3 -- whether any row carries a route contract.
1618pub(crate) fn route_active(table: &[MonitorSpec]) -> bool {
1619    table.iter().any(|m| m.route_max_latency_ms > 0)
1620}
1621
1622/// phase10-B3 -- drop every waiting take and the report-once state (arming:
1623/// a take from start-up is not a route of RUN's).
1624pub(crate) fn route_reset(table: &[MonitorSpec]) {
1625    for spec in table.iter() {
1626        let w = spec.cell.route.load(Ordering::Relaxed);
1627        spec.cell
1628            .route
1629            .store(w & ROUTE_SLOT_MASK, Ordering::Relaxed);
1630    }
1631}
1632
1633/// Per-age-spec dedup state.
1634#[derive(Debug, Clone, Copy, Default)]
1635pub(crate) struct AgeState {
1636    pub(crate) violated_last_window: bool,
1637    /// phase-462 W2 -- the silence rule's window. `opened` is a plain bool for
1638    /// the same reason [`MonitorState`]'s is: `now_us == 0` is a legitimate
1639    /// first sample.
1640    pub(crate) silence_opened: bool,
1641    /// Report-once-until-recovery, matching every other rule here.
1642    pub(crate) silence_reported: bool,
1643    /// Monotonic MILLISECONDS of the last check that saw a take (or of the
1644    /// first check, which starts the lease rather than judging it).
1645    ///
1646    /// Milliseconds and `u32`, not the uss the clock hands out, because this
1647    /// array is INLINE in the `Executor` value (`[AgeState; MAX_AGE_MONITORS]`) and
1648    /// that value has a byte budget a knob-scaled table must not blow (issue
1649    /// 0961 / `storage.rs`'s `the_executor_value_does_not_scale_with_the_knobs`
1650    /// -- a `u64` here costs 8 rows x 16 B and fails it). The unit is the
1651    /// contract's own (`max_age_ms`), the comparison is a wrapping delta, so
1652    /// the 49-day rollover is not a discontinuity, and a silence window is
1653    /// never shorter than a millisecond.
1654    pub(crate) last_take_ms: u32,
1655}
1656
1657/// phase-462 W2 -- the on-target liveliness lease: a contracted subscription
1658/// that has taken NOTHING for a whole `max_age_ms` window.
1659///
1660/// The bound needs no new declaration and deliberately does not invent one.
1661/// `max_age_ms` already says how old this input's data may be; data that never
1662/// arrives is older than that by the same clock, so the window IS the declared
1663/// bound. (`on_violation` selects what the executor then DOES about it; the
1664/// reaction beyond reporting is the fault hook phase-462 W3 owns.)
1665///
1666/// Why this is not `max-age-runtime` with a longer arm: that rule judges a
1667/// message that arrived. An input that stops entirely produces no message to
1668/// judge, so it leaves the age rule silent forever -- the exact failure a
1669/// `max_age_ms` promise is made against, and the one play_launch's reaction
1670/// engine catches on the Linux side of the same contract.
1671///
1672/// `now_us` is the executor's monotonic clock, `None` on a build with no clock
1673/// at all. Then nothing is judged: a clock that never moves is a window that
1674/// never elapses, and guessing a window from spin counts would put a tolerance
1675/// nobody declared into a safety rule.
1676///
1677/// phase-462 W5b -- `window_ms` is [`silence_window`]'s: the age bound, or the
1678/// contract's `deadline` when the row is armed for one and it is shorter.
1679fn check_silence(window_ms: u32, state: &mut AgeState, took: u32, now_us: u64) -> bool {
1680    let now_ms = (now_us / 1_000) as u32;
1681    if !state.silence_opened {
1682        state.silence_opened = true;
1683        state.last_take_ms = now_ms;
1684        return false;
1685    }
1686    if took > 0 {
1687        state.last_take_ms = now_ms;
1688        state.silence_reported = false;
1689        return false;
1690    }
1691    if now_ms.wrapping_sub(state.last_take_ms) < window_ms {
1692        return false;
1693    }
1694    if state.silence_reported {
1695        return false; // still silent -- already reported
1696    }
1697    state.silence_reported = true;
1698    true
1699}
1700
1701/// Pure age check: drains the sub cell's window-max take-age and fires
1702/// when it exceeds the declared bound. Report-once-until-recovery.
1703///
1704/// phase-462 W2 -- also the silence rule's tick, because the two verdicts read
1705/// ONE observation (what this endpoint took since the last check) and reading
1706/// it twice would let one rule's drain hide the other's evidence. They cannot
1707/// both fire: with no takes the drained age is 0, so `max-age-runtime` has
1708/// nothing to judge.
1709/// phase-462 W5b -- the silence rule's window and its evidence for one row:
1710/// `(window_ms, evidence)`, or `None` when the row judges no silence at all.
1711///
1712/// An age contract keeps W2's evidence (stamped takes, so an endpoint whose
1713/// type carries no stamp still reads as unjudgeable) and an armed deadline
1714/// shortens the window when it is the tighter bound. A row armed for a
1715/// deadline ALONE counts arrivals, since a deadline promises arrival and
1716/// nothing about stamps. Both counters are drained either way, so neither
1717/// carries a stale window into the next check.
1718fn silence_window(spec: &AgeMonitorSpec) -> Option<(u32, u32)> {
1719    let took = spec.cell.takes.swap(0, Ordering::Relaxed);
1720    let arrived = spec.cell.arrivals.swap(0, Ordering::Relaxed);
1721    let armed = spec.deadline_ms > 0 && spec.cell.deadline_slot.load(Ordering::Relaxed) != 0;
1722    match (spec.max_age_ms, armed) {
1723        (0, false) => None,
1724        (0, true) => Some((spec.deadline_ms, arrived)),
1725        (age, false) => Some((age, took)),
1726        (age, true) => Some((age.min(spec.deadline_ms), took)),
1727    }
1728}
1729
1730pub(crate) fn check_age(
1731    spec: &AgeMonitorSpec,
1732    state: &mut AgeState,
1733    now_us: Option<u64>,
1734) -> Option<Violation> {
1735    let (window_ms, took) = silence_window(spec)?;
1736    if let Some(now_us) = now_us
1737        && check_silence(window_ms, state, took, now_us)
1738    {
1739        return Some(Violation {
1740            rule: "silence-runtime",
1741            fqn: spec.fqn,
1742            // Takes in the window, against the window the contract declares.
1743            measured: 0,
1744            declared: window_ms,
1745        });
1746    }
1747    if spec.max_age_ms == 0 {
1748        return None;
1749    }
1750    let max_ms = spec.cell.max_age_ms.swap(0, Ordering::Relaxed);
1751    if max_ms > spec.max_age_ms {
1752        if state.violated_last_window {
1753            return None;
1754        }
1755        state.violated_last_window = true;
1756        Some(Violation {
1757            rule: "max-age-runtime",
1758            fqn: spec.fqn,
1759            measured: max_ms,
1760            declared: spec.max_age_ms,
1761        })
1762    } else {
1763        state.violated_last_window = false;
1764        None
1765    }
1766}
1767
1768// ---------------------------------------------------------------------------
1769// phase-462 W3b -- the hazard observer.
1770// ---------------------------------------------------------------------------
1771
1772/// One guard member: which monitor row's verdict says this topic has faulted.
1773///
1774/// The observer adds no detector of its own. A guard topic has faulted when
1775/// the row the image already bakes for it says so: an age row whose
1776/// `silence-runtime` lease is reported, or a rate row whose last window was
1777/// short (`min-rate-runtime`). Reading the rows' state rather than their
1778/// violations keeps the two in step by construction -- a hazard cannot be
1779/// detected by a rule that did not fire, nor stay detected after it recovered.
1780#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1781pub enum HazardMonitorRef {
1782    /// Row index into the executor's age table.
1783    Age(u16),
1784    /// Row index into the executor's rate/latency table.
1785    Rate(u16),
1786}
1787
1788/// One fault a hazard covers: a topic, or a redundant set of topics.
1789#[derive(Debug, Clone, Copy)]
1790pub struct HazardMonitorSpec {
1791    /// `true`: the guard faults only when EVERY member has (a redundant
1792    /// set). `false`: when any member has.
1793    pub all_of: bool,
1794    /// The guard's members.
1795    pub members: &'static [HazardMonitorRef],
1796}
1797
1798/// [`HazardSpec::sink`] when the hazard's reaction states no `safe_state`
1799/// publisher this image monitors: the observer reports detection and
1800/// recovery, and no reaction.
1801pub const NO_SINK: u16 = u16::MAX;
1802
1803/// One hazard's observer state. Baked as a `static` beside the spec, the way
1804/// [`PubMonitorCell`] is, so the observer adds a slice to the `Executor`
1805/// value and nothing per hazard (issue 0961).
1806#[derive(Debug, Default)]
1807pub struct HazardCell {
1808    /// 0 clear, 1 detected, 2 detected and reacted.
1809    state: AtomicU32,
1810    /// Monotonic ms the fault began: the faulting age row's last take, or the
1811    /// tick that first saw a rate guard short.
1812    fault_ms: AtomicU32,
1813    /// The sink publisher's count when the fault was detected.
1814    sink_count: AtomicU32,
1815    /// Index into [`HazardSpec::rungs`] of the rung being judged.
1816    rung_active: AtomicU32,
1817    /// Monotonic ms the active rung became active: the detection tick (the
1818    /// fire), or the tick that saw the ladder fall to it.
1819    rung_armed_ms: AtomicU32,
1820    /// Monotonic ms the active rung's route clock started: its `starts_at`
1821    /// node's detection when this executor observes that node, else
1822    /// [`Self::rung_armed_ms`].
1823    rung_clock_ms: AtomicU32,
1824    /// The active rung's output publisher count when its clock started.
1825    rung_count: AtomicU32,
1826    /// [`RUNG_IDLE`], [`RUNG_ARMED`], [`RUNG_WATCHING`] or [`RUNG_JUDGED`].
1827    rung_state: AtomicU32,
1828}
1829
1830impl HazardCell {
1831    /// A cleared cell, for a `static`.
1832    pub const fn new() -> Self {
1833        Self {
1834            state: AtomicU32::new(0),
1835            fault_ms: AtomicU32::new(0),
1836            sink_count: AtomicU32::new(0),
1837            rung_active: AtomicU32::new(0),
1838            rung_armed_ms: AtomicU32::new(0),
1839            rung_clock_ms: AtomicU32::new(0),
1840            rung_count: AtomicU32::new(0),
1841            rung_state: AtomicU32::new(RUNG_IDLE),
1842        }
1843    }
1844}
1845
1846/// No rung route being watched.
1847const RUNG_IDLE: u32 = 0;
1848/// A rung is active; its route clock waits for its `starts_at` detection.
1849const RUNG_ARMED: u32 = 1;
1850/// The route clock runs; waiting for the rung's first output.
1851const RUNG_WATCHING: u32 = 2;
1852/// The active rung was judged (answered, or reported overdue).
1853const RUNG_JUDGED: u32 = 3;
1854
1855/// [`RungSpec::mode`] for a rung that is not a contract mode (a reaction that
1856/// is a single path has one rung, named after the path).
1857pub const NO_MODE: u16 = u16::MAX;
1858
1859/// [`RungSpec::role`] values, in rlm's `LevelRole` order (~~`RungRole`~~ before rlm v0.3.3).
1860pub const RUNG_ROLE_RUNG: u8 = 0;
1861/// A `window:` rung: lasts at least its window; the rung below takes over
1862/// when the window passes.
1863pub const RUNG_ROLE_WINDOW: u8 = 1;
1864/// The ladder's last rung.
1865pub const RUNG_ROLE_FLOOR: u8 = 2;
1866
1867/// play_launch phase 96 W4.2 (phase 85 D4) -- one rung of a hazard's ladder
1868/// (`contracts.hazards.<h>.rungs[]`, top first), lowered by the bake.
1869#[derive(Debug, Clone, Copy)]
1870pub struct RungSpec {
1871    /// The baked mode id the rung IS (the id `request_mode` takes), or
1872    /// [`NO_MODE`]. While the executor is in this mode, the ladder has fallen
1873    /// to this rung.
1874    pub mode: u16,
1875    /// Rate-table row of a publisher on one of the rung's `output` topics, or
1876    /// [`NO_SINK`] when this executor does not observe one.
1877    pub output_row: u16,
1878    /// The rung's ROUTE as play_launch's checker charged it (`route_ms`,
1879    /// rounded up): from its `starts_at` node's detection to the rung's first
1880    /// output, excluding the detection time, any window and the settle. 0 = no
1881    /// route, so the rung is tracked but not judged.
1882    pub route_ms: u32,
1883    /// [`RUNG_ROLE_RUNG`], [`RUNG_ROLE_WINDOW`] or [`RUNG_ROLE_FLOOR`].
1884    pub role: u8,
1885    /// The hazard's guard members that belong to the rung's `starts_at` node:
1886    /// the route clock starts when one of them reports faulted. Empty when
1887    /// this executor does not observe that node as a detector (a rung below a
1888    /// window starts at the window owner, which is not a guard), and then the
1889    /// clock starts when the rung becomes active.
1890    pub starts_at: &'static [HazardMonitorRef],
1891}
1892
1893/// One contracted hazard (`contracts.hazards`), lowered by the bake.
1894#[derive(Debug, Clone, Copy)]
1895pub struct HazardSpec {
1896    /// The hazard's contract key (`<scope>/<name>`), the report key.
1897    pub key: &'static str,
1898    /// Declared `ftti_ms`; 0 = none declared.
1899    pub ftti_ms: u32,
1900    /// The reacting path's `safe_state.settle_ms`, added to the reaction
1901    /// time as play_launch adds it; 0 = none.
1902    pub settle_ms: u32,
1903    /// The faults the hazard covers; any one of them is the hazard.
1904    pub monitors: &'static [HazardMonitorSpec],
1905    /// Rate-table row of the `safe_state.emits` publisher, or [`NO_SINK`].
1906    pub sink: u16,
1907    /// phase-462 W3c -- the id the reaction hook receives when this hazard is
1908    /// DETECTED (the reacting path's, in the model's one reaction id space
1909    /// shared with supervised routes); 0 = no safe-state entry to start.
1910    pub reaction_id: u16,
1911    /// play_launch phase 96 W4.2 (phase 85 D4) -- the hazard's ladder, top
1912    /// rung first. Empty: no rung route is monitored.
1913    pub rungs: &'static [RungSpec],
1914    /// The observer's state.
1915    pub cell: &'static HazardCell,
1916}
1917
1918const HAZARD_CLEAR: u32 = 0;
1919const HAZARD_DETECTED: u32 = 1;
1920const HAZARD_REACTED: u32 = 2;
1921
1922/// Whether `member` has faulted, and since when (ms) when the row knows.
1923fn member_fault(
1924    member: HazardMonitorRef,
1925    monitor_states: &[MonitorState],
1926    age_states: &[AgeState],
1927) -> Option<Option<u32>> {
1928    match member {
1929        HazardMonitorRef::Age(i) => {
1930            let s = age_states.get(i as usize)?;
1931            s.silence_reported.then_some(Some(s.last_take_ms))
1932        }
1933        HazardMonitorRef::Rate(i) => {
1934            let s = monitor_states.get(i as usize)?;
1935            s.violated_last_window.then_some(None)
1936        }
1937    }
1938}
1939
1940/// Whether any guard of `spec` has faulted, and the earliest fault start the
1941/// faulting rows know.
1942fn hazard_fault(
1943    spec: &HazardSpec,
1944    monitor_states: &[MonitorState],
1945    age_states: &[AgeState],
1946) -> Option<Option<u32>> {
1947    let mut faulted = false;
1948    let mut since: Option<u32> = None;
1949    for g in spec.monitors {
1950        let mut seen = 0usize;
1951        let mut g_since: Option<u32> = None;
1952        for m in g.members {
1953            if let Some(at) = member_fault(*m, monitor_states, age_states) {
1954                seen += 1;
1955                if let Some(at) = at {
1956                    g_since = Some(g_since.map_or(at, |s: u32| s.min(at)));
1957                }
1958            }
1959        }
1960        let guard_faulted = if g.all_of {
1961            !g.members.is_empty() && seen == g.members.len()
1962        } else {
1963            seen > 0
1964        };
1965        if guard_faulted {
1966            faulted = true;
1967            if let Some(at) = g_since {
1968                since = Some(since.map_or(at, |s| s.min(at)));
1969            }
1970        }
1971    }
1972    faulted.then_some(since)
1973}
1974
1975/// phase-462 W3b -- one tick of a hazard's observer, after the monitor rows
1976/// have run. Emits, in play_launch's vocabulary so one log reader serves both
1977/// sides of a bridge:
1978///
1979/// - `hazard-detected` when a guard first faults: measured is the ms since
1980///   the fault began, declared the FTTI;
1981/// - `hazard-reaction` when the `safe_state` publisher publishes after that:
1982///   measured is fault-to-reaction plus `settle_ms`, declared the FTTI;
1983/// - `hazard-recovered` when no guard is faulted any more: measured is the
1984///   ms the hazard lasted;
1985/// - `hazard-route-runtime` (play_launch phase 96 W4.2) when the ACTIVE
1986///   rung's ROUTE overran: measured is the ms from the rung's route clock
1987///   start to the first publish on its output, or, when none came, to the
1988///   tick that saw the route pass; declared is the rung's `route_ms`. A rung
1989///   that answers within its route reports nothing. Each rung that becomes
1990///   active is judged at most once.
1991///
1992/// Which rung is active comes from `mode`, the mode the executor judges by:
1993/// at detection it is the rung whose mode the executor is already in (the
1994/// ladder fell before this fault), else the top rung; while the hazard holds,
1995/// the executor entering a LOWER rung's mode is a fall, and that rung becomes
1996/// active. A rung's clock starts at its `starts_at` node's detection when this
1997/// executor observes that node among the guard's detectors, else when the
1998/// rung becomes active.
1999///
2000/// `fqn` is the hazard's key on all of them.
2001pub(crate) fn check_hazard(
2002    spec: &HazardSpec,
2003    monitor_table: &[MonitorSpec],
2004    monitor_states: &[MonitorState],
2005    age_states: &[AgeState],
2006    mode: u16,
2007    now_ms: u32,
2008    mut emit: impl FnMut(Violation),
2009) {
2010    let cell = spec.cell;
2011    let state = cell.state.load(Ordering::Relaxed);
2012    let sink_count = monitor_table
2013        .get(spec.sink as usize)
2014        .map(|m| m.cell.count.load(Ordering::Relaxed));
2015    let fault = hazard_fault(spec, monitor_states, age_states);
2016    let v = |rule, measured: u32| Violation {
2017        rule,
2018        fqn: spec.key,
2019        measured,
2020        declared: spec.ftti_ms,
2021    };
2022    match (state, fault) {
2023        (HAZARD_CLEAR, Some(since)) => {
2024            let at = since.unwrap_or(now_ms);
2025            cell.fault_ms.store(at, Ordering::Relaxed);
2026            cell.sink_count
2027                .store(sink_count.unwrap_or(0), Ordering::Relaxed);
2028            cell.state.store(HAZARD_DETECTED, Ordering::Relaxed);
2029            let active = spec
2030                .rungs
2031                .iter()
2032                .position(|r| r.mode != NO_MODE && r.mode == mode)
2033                .unwrap_or(0);
2034            arm_rung(spec, active, now_ms);
2035            emit(v("hazard-detected", now_ms.wrapping_sub(at)));
2036        }
2037        (HAZARD_DETECTED, Some(_)) => {
2038            if let Some(count) = sink_count
2039                && count != cell.sink_count.load(Ordering::Relaxed)
2040            {
2041                cell.state.store(HAZARD_REACTED, Ordering::Relaxed);
2042                let since = now_ms.wrapping_sub(cell.fault_ms.load(Ordering::Relaxed));
2043                emit(v("hazard-reaction", since.saturating_add(spec.settle_ms)));
2044            }
2045        }
2046        (HAZARD_DETECTED | HAZARD_REACTED, None) => {
2047            cell.state.store(HAZARD_CLEAR, Ordering::Relaxed);
2048            cell.rung_state.store(RUNG_IDLE, Ordering::Relaxed);
2049            let lasted = now_ms.wrapping_sub(cell.fault_ms.load(Ordering::Relaxed));
2050            emit(v("hazard-recovered", lasted));
2051        }
2052        _ => {}
2053    }
2054    if fault.is_some() {
2055        check_rung(
2056            spec,
2057            monitor_table,
2058            monitor_states,
2059            age_states,
2060            mode,
2061            now_ms,
2062            &mut emit,
2063        );
2064    }
2065}
2066
2067/// Make rung `active` the one being judged, from `now_ms`.
2068fn arm_rung(spec: &HazardSpec, active: usize, now_ms: u32) {
2069    let cell = spec.cell;
2070    if active >= spec.rungs.len() {
2071        cell.rung_state.store(RUNG_IDLE, Ordering::Relaxed);
2072        return;
2073    }
2074    cell.rung_active.store(active as u32, Ordering::Relaxed);
2075    cell.rung_armed_ms.store(now_ms, Ordering::Relaxed);
2076    cell.rung_state.store(RUNG_ARMED, Ordering::Relaxed);
2077}
2078
2079/// play_launch phase 96 W4.2 -- while the hazard holds: follow a fall to a
2080/// lower rung, start the active rung's route clock at its `starts_at`
2081/// detection, and judge the rung once, at its first output or when its route
2082/// passes with none.
2083fn check_rung(
2084    spec: &HazardSpec,
2085    monitor_table: &[MonitorSpec],
2086    monitor_states: &[MonitorState],
2087    age_states: &[AgeState],
2088    mode: u16,
2089    now_ms: u32,
2090    emit: &mut impl FnMut(Violation),
2091) {
2092    let cell = spec.cell;
2093    if cell.rung_state.load(Ordering::Relaxed) == RUNG_IDLE {
2094        return;
2095    }
2096    // A fall: the executor entered the mode of a rung below the active one.
2097    let active = cell.rung_active.load(Ordering::Relaxed) as usize;
2098    if let Some(lower) = spec
2099        .rungs
2100        .iter()
2101        .enumerate()
2102        .skip(active + 1)
2103        .find(|(_, r)| r.mode != NO_MODE && r.mode == mode)
2104        .map(|(i, _)| i)
2105    {
2106        arm_rung(spec, lower, now_ms);
2107    }
2108    let active = cell.rung_active.load(Ordering::Relaxed) as usize;
2109    let Some(rung) = spec.rungs.get(active) else {
2110        return;
2111    };
2112    if rung.route_ms == 0 {
2113        return;
2114    }
2115    let Some(count) = monitor_table
2116        .get(rung.output_row as usize)
2117        .map(|m| m.cell.count.load(Ordering::Relaxed))
2118    else {
2119        return;
2120    };
2121    match cell.rung_state.load(Ordering::Relaxed) {
2122        RUNG_ARMED => {
2123            let started = rung.starts_at.is_empty()
2124                || rung
2125                    .starts_at
2126                    .iter()
2127                    .any(|m| member_fault(*m, monitor_states, age_states).is_some());
2128            if !started {
2129                return;
2130            }
2131            let clock = if rung.starts_at.is_empty() {
2132                cell.rung_armed_ms.load(Ordering::Relaxed)
2133            } else {
2134                now_ms
2135            };
2136            cell.rung_clock_ms.store(clock, Ordering::Relaxed);
2137            cell.rung_count.store(count, Ordering::Relaxed);
2138            cell.rung_state.store(RUNG_WATCHING, Ordering::Relaxed);
2139        }
2140        RUNG_WATCHING => {
2141            let elapsed = now_ms.wrapping_sub(cell.rung_clock_ms.load(Ordering::Relaxed));
2142            let answered = count != cell.rung_count.load(Ordering::Relaxed);
2143            if !answered && elapsed <= rung.route_ms {
2144                return;
2145            }
2146            cell.rung_state.store(RUNG_JUDGED, Ordering::Relaxed);
2147            if elapsed > rung.route_ms {
2148                emit(Violation {
2149                    rule: "hazard-route-runtime",
2150                    fqn: spec.key,
2151                    measured: elapsed,
2152                    declared: rung.route_ms,
2153                });
2154            }
2155        }
2156        _ => {}
2157    }
2158}
2159
2160#[cfg(test)]
2161mod hazard_tests {
2162    use super::*;
2163
2164    static PUB_CELL: PubMonitorCell = PubMonitorCell::new();
2165    static SUB_CELL: SubMonitorCell = SubMonitorCell::new();
2166    static SINK_CELL: PubMonitorCell = PubMonitorCell::new();
2167
2168    static RATE_ROWS: [MonitorSpec; 2] = [
2169        MonitorSpec {
2170            topic: "/scan",
2171            fqn: "/sensor/scan",
2172            min_rate_hz_milli: 10_000,
2173            max_latency_ms: 0,
2174            route_trigger: "",
2175            route_max_latency_ms: 0,
2176            route_deadline_ms: 0,
2177            reaction_id: 0,
2178            mechanism: 0,
2179            cell: &PUB_CELL,
2180        },
2181        MonitorSpec {
2182            topic: "/stop",
2183            fqn: "/handler/stop",
2184            min_rate_hz_milli: 0,
2185            max_latency_ms: 0,
2186            route_trigger: "",
2187            route_max_latency_ms: 0,
2188            route_deadline_ms: 0,
2189            reaction_id: 0,
2190            mechanism: 0,
2191            cell: &SINK_CELL,
2192        },
2193    ];
2194    #[allow(dead_code)]
2195    static AGE_ROW: AgeMonitorSpec = AgeMonitorSpec {
2196        topic: "/scan",
2197        fqn: "/handler/scan",
2198        max_age_ms: 100,
2199        deadline_ms: 0,
2200        cell: &SUB_CELL,
2201    };
2202
2203    fn run(
2204        spec: &HazardSpec,
2205        mon: &[MonitorState],
2206        age: &[AgeState],
2207        now: u32,
2208    ) -> std::vec::Vec<Violation> {
2209        let mut out = std::vec::Vec::new();
2210        check_hazard(spec, &RATE_ROWS, mon, age, 0, now, |v| out.push(v));
2211        out
2212    }
2213
2214    /// detected -> reaction -> recovered, each once, on a silence guard, with
2215    /// times measured from the guard's last take.
2216    #[test]
2217    fn a_silence_guard_detects_reacts_and_recovers_once_each() {
2218        static CELL: HazardCell = HazardCell::new();
2219        static MONITORS: [HazardMonitorSpec; 1] = [HazardMonitorSpec {
2220            all_of: false,
2221            members: &[HazardMonitorRef::Age(0)],
2222        }];
2223        let spec = HazardSpec {
2224            key: "demo.launch.xml/scan_lost",
2225            ftti_ms: 500,
2226            settle_ms: 30,
2227            monitors: &MONITORS,
2228            sink: 1,
2229            reaction_id: 0,
2230            rungs: &[],
2231            cell: &CELL,
2232        };
2233        let mon = [MonitorState::default(); 2];
2234        let mut age = [AgeState::default(); 1];
2235        assert!(run(&spec, &mon, &age, 1_000).is_empty(), "clear: nothing");
2236
2237        age[0].silence_reported = true;
2238        age[0].last_take_ms = 900;
2239        let v = run(&spec, &mon, &age, 1_000);
2240        assert_eq!(v.len(), 1);
2241        assert_eq!((v[0].rule, v[0].fqn), ("hazard-detected", spec.key));
2242        assert_eq!((v[0].measured, v[0].declared), (100, 500));
2243        assert!(run(&spec, &mon, &age, 1_010).is_empty(), "detected once");
2244
2245        SINK_CELL.count.fetch_add(1, Ordering::Relaxed);
2246        let v = run(&spec, &mon, &age, 1_050);
2247        assert_eq!(v.len(), 1);
2248        assert_eq!(v[0].rule, "hazard-reaction");
2249        assert_eq!(v[0].measured, 150 + 30, "fault-to-reaction plus settle");
2250        SINK_CELL.count.fetch_add(1, Ordering::Relaxed);
2251        assert!(run(&spec, &mon, &age, 1_060).is_empty(), "reacted once");
2252
2253        age[0].silence_reported = false;
2254        let v = run(&spec, &mon, &age, 1_200);
2255        assert_eq!(v.len(), 1);
2256        assert_eq!((v[0].rule, v[0].measured), ("hazard-recovered", 300));
2257        assert!(run(&spec, &mon, &age, 1_210).is_empty(), "recovered once");
2258    }
2259
2260    /// A redundant set faults only when every member has.
2261    #[test]
2262    fn an_all_of_guard_waits_for_every_member() {
2263        static CELL: HazardCell = HazardCell::new();
2264        static MONITORS: [HazardMonitorSpec; 1] = [HazardMonitorSpec {
2265            all_of: true,
2266            members: &[HazardMonitorRef::Age(0), HazardMonitorRef::Rate(0)],
2267        }];
2268        let spec = HazardSpec {
2269            key: "demo.launch.xml/both_lost",
2270            ftti_ms: 0,
2271            settle_ms: 0,
2272            monitors: &MONITORS,
2273            sink: NO_SINK,
2274            reaction_id: 0,
2275            rungs: &[],
2276            cell: &CELL,
2277        };
2278        let mut mon = [MonitorState::default(); 2];
2279        let mut age = [AgeState::default(); 1];
2280        age[0].silence_reported = true;
2281        assert!(
2282            run(&spec, &mon, &age, 10).is_empty(),
2283            "one of two: no hazard"
2284        );
2285        mon[0].violated_last_window = true;
2286        let v = run(&spec, &mon, &age, 20);
2287        assert_eq!(v[0].rule, "hazard-detected");
2288        assert_eq!(v[0].declared, 0, "no FTTI declared");
2289    }
2290
2291    /// phase-462 W4 -- the text `mode_variants::render_mode_tables_rs` emits
2292    /// after the base tables for a one-row model with one mode (its
2293    /// `the_variant_tables_render_over_the_base_cells` fixture), verbatim but
2294    /// for the crate root, so the generated shape compiles against the real
2295    /// types. phase-462 W4c -- with the deadline rows and switch budget of its
2296    /// `the_variant_tables_render_deadlines_and_the_switch_budget` fixture.
2297    #[allow(dead_code, non_upper_case_globals)]
2298    mod rendered_modes {
2299        use crate::executor::monitor::{AgeMonitorSpec, MonitorSpec, PubMonitorCell};
2300
2301        static NROS_MONITOR_CELL_0: PubMonitorCell = PubMonitorCell::new();
2302        pub static NROS_MONITORS: &[MonitorSpec] = &[MonitorSpec {
2303            topic: "/cmd",
2304            fqn: "/talker/cmd",
2305            min_rate_hz_milli: 10000u32,
2306            max_latency_ms: 20u32,
2307            route_trigger: "",
2308            route_max_latency_ms: 0u32,
2309            route_deadline_ms: 0u32,
2310            reaction_id: 0u16,
2311            mechanism: 0u8,
2312            cell: &NROS_MONITOR_CELL_0,
2313        }];
2314        pub static NROS_AGE_MONITORS: &[AgeMonitorSpec] = &[];
2315
2316        // phase-462 W4 -- the contract's modes: the base rows, the base cells, the
2317        // mode's numbers.
2318        pub static NROS_MODE_MONITORS_1: &[MonitorSpec] = &[MonitorSpec {
2319            topic: "/cmd",
2320            fqn: "/talker/cmd",
2321            min_rate_hz_milli: 5000u32,
2322            max_latency_ms: 20u32,
2323            route_trigger: "",
2324            route_max_latency_ms: 0u32,
2325            route_deadline_ms: 0u32,
2326            reaction_id: 0u16,
2327            mechanism: 0u8,
2328            cell: &NROS_MONITOR_CELL_0,
2329        }];
2330        pub static NROS_MODE_AGE_MONITORS_1: &[AgeMonitorSpec] = &[];
2331        pub static NROS_MODE_DEADLINES_0: &[crate::executor::monitor::ModeDeadline] =
2332            &[crate::executor::monitor::ModeDeadline {
2333                sc: 0u8,
2334                deadline_us: 5000u32,
2335            }];
2336        pub static NROS_MODE_DEADLINES_1: &[crate::executor::monitor::ModeDeadline] =
2337            &[crate::executor::monitor::ModeDeadline {
2338                sc: 0u8,
2339                deadline_us: 2000u32,
2340            }];
2341        pub static NROS_MODE_TABLES: &[crate::executor::monitor::ModeTables] = &[
2342            crate::executor::monitor::ModeTables {
2343                monitors: NROS_MONITORS,
2344                ages: NROS_AGE_MONITORS,
2345                deadlines: NROS_MODE_DEADLINES_0,
2346                switch_budget_ms: 0u32,
2347                reaction: "",
2348            },
2349            crate::executor::monitor::ModeTables {
2350                monitors: NROS_MODE_MONITORS_1,
2351                ages: NROS_MODE_AGE_MONITORS_1,
2352                deadlines: NROS_MODE_DEADLINES_1,
2353                switch_budget_ms: 50u32,
2354                reaction: "//enter_safe",
2355            },
2356        ];
2357
2358        /// Install the mode variants AFTER `nros_install_monitors`; then
2359        /// `request_mode(id)` switches this executor at its next spin.
2360        pub fn nros_install_mode_tables(
2361            executor: &mut crate::executor::Executor<'_>,
2362        ) -> ::core::result::Result<(), crate::executor::monitor::ModeTablesMismatch> {
2363            executor.set_mode_tables(NROS_MODE_TABLES)
2364        }
2365    }
2366
2367    /// The text `contract_monitors::render_monitor_rs_full` emits for the
2368    /// W3a/W3b fixture (one sensor, one handler, one hazard), verbatim but for
2369    /// the crate root. It is here so the GENERATED shape is compiled against
2370    /// the real types: a renderer change that stops compiling has to change
2371    /// this copy too, in plain sight.
2372    #[allow(dead_code, non_upper_case_globals)]
2373    mod rendered {
2374        use crate::executor::monitor::{
2375            AgeMonitorSpec, MonitorSpec, PubMonitorCell, SubMonitorCell,
2376        };
2377
2378        static NROS_MONITOR_CELL_0: PubMonitorCell = PubMonitorCell::new();
2379        static NROS_MONITOR_CELL_1: PubMonitorCell = PubMonitorCell::new();
2380        static NROS_AGE_CELL_0: SubMonitorCell = SubMonitorCell::new();
2381
2382        pub static NROS_MONITORS: &[MonitorSpec] = &[
2383            MonitorSpec {
2384                topic: "/stop",
2385                fqn: "/handler/stop",
2386                min_rate_hz_milli: 0u32,
2387                max_latency_ms: 20u32,
2388                route_trigger: "/scan",
2389                route_max_latency_ms: 20u32,
2390                route_deadline_ms: 0u32,
2391                reaction_id: 0u16,
2392                mechanism: 0u8,
2393                cell: &NROS_MONITOR_CELL_0,
2394            },
2395            MonitorSpec {
2396                topic: "/scan",
2397                fqn: "/sensor/scan",
2398                min_rate_hz_milli: 10000u32,
2399                max_latency_ms: 0u32,
2400                route_trigger: "",
2401                route_max_latency_ms: 0u32,
2402                route_deadline_ms: 0u32,
2403                reaction_id: 0u16,
2404                mechanism: 0u8,
2405                cell: &NROS_MONITOR_CELL_1,
2406            },
2407        ];
2408
2409        pub static NROS_AGE_MONITORS: &[AgeMonitorSpec] = &[AgeMonitorSpec {
2410            topic: "/scan",
2411            fqn: "/handler/scan",
2412            max_age_ms: 100u32,
2413            deadline_ms: 0,
2414            cell: &NROS_AGE_CELL_0,
2415        }];
2416
2417        use crate::executor::monitor::{
2418            HazardCell, HazardMonitorRef, HazardMonitorSpec, HazardSpec,
2419        };
2420
2421        static NROS_HAZARD_CELL_0: HazardCell = HazardCell::new();
2422        const NROS_HAZARD_MONITORS_0: &[HazardMonitorSpec] = &[HazardMonitorSpec {
2423            all_of: false,
2424            members: &[HazardMonitorRef::Age(0u16), HazardMonitorRef::Rate(1u16)],
2425        }];
2426
2427        pub static NROS_HAZARDS: &[HazardSpec] = &[HazardSpec {
2428            key: "demo.launch.xml/scan_lost",
2429            ftti_ms: 0u32,
2430            settle_ms: 30u32,
2431            monitors: NROS_HAZARD_MONITORS_0,
2432            sink: 0u16,
2433            reaction_id: 1u16,
2434            rungs: &[crate::executor::monitor::RungSpec {
2435                mode: crate::executor::monitor::NO_MODE,
2436                output_row: 0u16,
2437                route_ms: 25u32,
2438                role: 2u8,
2439                starts_at: &[HazardMonitorRef::Age(0u16)],
2440            }],
2441            cell: &NROS_HAZARD_CELL_0,
2442        }];
2443    }
2444
2445    /// The rendered tables drive the observer: the age row's silence is the
2446    /// hazard, and the sink row the renderer chose is the one it watches.
2447    #[test]
2448    fn the_rendered_include_drives_the_observer() {
2449        let spec = &rendered::NROS_HAZARDS[0];
2450        let mon = [MonitorState::default(); 2];
2451        let mut age = [AgeState::default(); 1];
2452        age[0].silence_reported = true;
2453        age[0].last_take_ms = 40;
2454        let mut out = std::vec::Vec::new();
2455        check_hazard(spec, rendered::NROS_MONITORS, &mon, &age, 0, 100, |v| {
2456            out.push(v)
2457        });
2458        assert_eq!((out[0].rule, out[0].measured), ("hazard-detected", 60));
2459        assert_eq!(
2460            rendered::NROS_MONITORS[spec.sink as usize].fqn,
2461            "/handler/stop"
2462        );
2463    }
2464
2465    /// play_launch phase 96 W4.2 -- a rung's route runs from the detection
2466    /// tick (the fire), not from the fault: a rung that answers within
2467    /// `route_ms` reports nothing, one that answers late reports the elapsed
2468    /// time, one that never answers is reported once when its route passes,
2469    /// and each detection is judged once.
2470    #[test]
2471    fn a_rung_route_is_judged_from_the_fire_once_per_detection() {
2472        static CELL: HazardCell = HazardCell::new();
2473        static RUNG_CELL: PubMonitorCell = PubMonitorCell::new();
2474        static ROWS: [MonitorSpec; 1] = [MonitorSpec {
2475            topic: "/brake",
2476            fqn: "/handler/brake",
2477            min_rate_hz_milli: 0,
2478            max_latency_ms: 0,
2479            route_trigger: "",
2480            route_max_latency_ms: 0,
2481            route_deadline_ms: 0,
2482            reaction_id: 0,
2483            mechanism: 0,
2484            cell: &RUNG_CELL,
2485        }];
2486        static MONITORS: [HazardMonitorSpec; 1] = [HazardMonitorSpec {
2487            all_of: false,
2488            members: &[HazardMonitorRef::Age(0)],
2489        }];
2490        let spec = HazardSpec {
2491            key: "demo.launch.xml/scan_lost",
2492            ftti_ms: 500,
2493            settle_ms: 0,
2494            monitors: &MONITORS,
2495            sink: NO_SINK,
2496            reaction_id: 0,
2497            rungs: &[RungSpec {
2498                mode: NO_MODE,
2499                output_row: 0,
2500                route_ms: 20,
2501                role: RUNG_ROLE_FLOOR,
2502                starts_at: &[],
2503            }],
2504            cell: &CELL,
2505        };
2506        let mon = [MonitorState::default(); 1];
2507        let mut age = [AgeState::default(); 1];
2508        let tick = |age: &[AgeState], now: u32| {
2509            let mut out = std::vec::Vec::new();
2510            check_hazard(&spec, &ROWS, &mon, age, 0, now, |v| out.push(v));
2511            out
2512        };
2513        let routes = |v: &[Violation]| -> std::vec::Vec<(u32, u32)> {
2514            v.iter()
2515                .filter(|v| v.rule == "hazard-route-runtime")
2516                .map(|v| (v.measured, v.declared))
2517                .collect()
2518        };
2519        let detect = |age: &mut [AgeState], now: u32| {
2520            age[0].silence_reported = true;
2521            age[0].last_take_ms = now - 150; // the fault began long before
2522            let v = tick(age, now);
2523            assert_eq!(v[0].rule, "hazard-detected");
2524            assert!(routes(&v).is_empty(), "nothing is judged at the fire");
2525        };
2526        let recover = |age: &mut [AgeState], now: u32| {
2527            age[0].silence_reported = false;
2528            let v = tick(age, now);
2529            assert_eq!(v[0].rule, "hazard-recovered");
2530        };
2531
2532        // Within: a publish 15 ms after the fire, route 20 ms.
2533        detect(&mut age, 1_000);
2534        assert!(routes(&tick(&age, 1_010)).is_empty(), "still within");
2535        RUNG_CELL.count.fetch_add(1, Ordering::Relaxed);
2536        assert!(routes(&tick(&age, 1_015)).is_empty(), "answered in time");
2537        RUNG_CELL.count.fetch_add(1, Ordering::Relaxed);
2538        assert!(routes(&tick(&age, 1_100)).is_empty(), "judged once");
2539        recover(&mut age, 1_200);
2540
2541        // Late: the first publish is seen at the tick 30 ms after the fire,
2542        // and 30 > 20. (The monitor has tick resolution, like
2543        // `hazard-reaction`.)
2544        detect(&mut age, 2_000);
2545        RUNG_CELL.count.fetch_add(1, Ordering::Relaxed);
2546        let v = tick(&age, 2_030);
2547        assert_eq!(routes(&v), [(30, 20)], "late: measured from the fire");
2548        RUNG_CELL.count.fetch_add(1, Ordering::Relaxed);
2549        assert!(routes(&tick(&age, 2_040)).is_empty(), "judged once");
2550        recover(&mut age, 2_100);
2551
2552        // Never: no publish; reported at the first tick past the route.
2553        detect(&mut age, 3_000);
2554        assert!(routes(&tick(&age, 3_020)).is_empty(), "20 is not past 20");
2555        assert_eq!(routes(&tick(&age, 3_021)), [(21, 20)], "overdue");
2556        assert!(routes(&tick(&age, 3_500)).is_empty(), "reported once");
2557        RUNG_CELL.count.fetch_add(1, Ordering::Relaxed);
2558        assert!(
2559            routes(&tick(&age, 3_510)).is_empty(),
2560            "a late answer after the overdue report is not a second verdict"
2561        );
2562        recover(&mut age, 3_600);
2563
2564        // A recovery before the rung answers ends the watch silently.
2565        detect(&mut age, 4_000);
2566        recover(&mut age, 4_005);
2567        assert!(
2568            routes(&tick(&age, 4_100)).is_empty(),
2569            "nothing after recovery"
2570        );
2571    }
2572
2573    /// A rung with no route (`route_ms` 0) never reports one, whatever
2574    /// `output_row` names: row 0 is a real row.
2575    #[test]
2576    fn no_route_no_route_verdict() {
2577        static CELL: HazardCell = HazardCell::new();
2578        static MONITORS: [HazardMonitorSpec; 1] = [HazardMonitorSpec {
2579            all_of: false,
2580            members: &[HazardMonitorRef::Rate(0)],
2581        }];
2582        let spec = HazardSpec {
2583            key: "k",
2584            ftti_ms: 100,
2585            settle_ms: 0,
2586            monitors: &MONITORS,
2587            sink: NO_SINK,
2588            reaction_id: 0,
2589            rungs: &[RungSpec {
2590                mode: NO_MODE,
2591                output_row: 0,
2592                route_ms: 0,
2593                role: RUNG_ROLE_FLOOR,
2594                starts_at: &[],
2595            }],
2596            cell: &CELL,
2597        };
2598        let mut mon = [MonitorState::default(); 2];
2599        mon[0].violated_last_window = true;
2600        assert_eq!(run(&spec, &mon, &[], 50)[0].rule, "hazard-detected");
2601        assert!(run(&spec, &mon, &[], 5_000).is_empty());
2602    }
2603
2604    /// A three-rung ladder -- a `window:` rung (mode 1), a rung (mode 2) and
2605    /// the floor (mode 3) -- each on its own output row, with distinct routes
2606    /// so a verdict names the rung it judged.
2607    ///
2608    /// Each test builds its OWN output cells and hazard cell: `cargo test`
2609    /// runs tests on parallel threads, and one shared `static` row let a
2610    /// `publish(0)` in one test answer the top rung in another, which then
2611    /// reported no route violation (seen as an intermittent failure of
2612    /// `a_fault_at_the_top_is_judged_against_the_top_rung`).
2613    mod ladder {
2614        use super::*;
2615
2616        const fn row(topic: &'static str, cell: &'static PubMonitorCell) -> MonitorSpec {
2617            MonitorSpec {
2618                topic,
2619                fqn: topic,
2620                min_rate_hz_milli: 0,
2621                max_latency_ms: 0,
2622                route_trigger: "",
2623                route_max_latency_ms: 0,
2624                route_deadline_ms: 0,
2625                reaction_id: 0,
2626                mechanism: 0,
2627                cell,
2628            }
2629        }
2630        pub static MONITORS: [HazardMonitorSpec; 1] = [HazardMonitorSpec {
2631            all_of: false,
2632            members: &[HazardMonitorRef::Age(0), HazardMonitorRef::Age(1)],
2633        }];
2634        /// The top rung starts at the detector the guard's age row 0 belongs
2635        /// to; the two below start at the window owner, which is not a guard
2636        /// detector here, so their clocks start when the ladder falls.
2637        pub static RUNGS: [RungSpec; 3] = [
2638            RungSpec {
2639                mode: 1,
2640                output_row: 0,
2641                route_ms: 10,
2642                role: RUNG_ROLE_WINDOW,
2643                starts_at: &[HazardMonitorRef::Age(0)],
2644            },
2645            RungSpec {
2646                mode: 2,
2647                output_row: 1,
2648                route_ms: 20,
2649                role: RUNG_ROLE_RUNG,
2650                starts_at: &[],
2651            },
2652            RungSpec {
2653                mode: 3,
2654                output_row: 2,
2655                route_ms: 30,
2656                role: RUNG_ROLE_FLOOR,
2657                starts_at: &[],
2658            },
2659        ];
2660        fn spec(cell: &'static HazardCell) -> HazardSpec {
2661            HazardSpec {
2662                key: "demo.launch.xml/scan_lost",
2663                ftti_ms: 0,
2664                settle_ms: 0,
2665                monitors: &MONITORS,
2666                sink: NO_SINK,
2667                reaction_id: 0,
2668                rungs: &RUNGS,
2669                cell,
2670            }
2671        }
2672        /// One test's ladder: its own output rows and hazard cell, leaked so
2673        /// they are `'static` like the generated tables they stand in for.
2674        pub struct Ladder {
2675            out: &'static [PubMonitorCell; 3],
2676            rows: &'static [MonitorSpec; 3],
2677            spec: HazardSpec,
2678        }
2679        impl Ladder {
2680            pub fn new() -> Self {
2681                let out: &'static [PubMonitorCell; 3] =
2682                    std::boxed::Box::leak(std::boxed::Box::new([
2683                        PubMonitorCell::new(),
2684                        PubMonitorCell::new(),
2685                        PubMonitorCell::new(),
2686                    ]));
2687                let rows: &'static [MonitorSpec; 3] =
2688                    std::boxed::Box::leak(std::boxed::Box::new([
2689                        row("/slow", &out[0]),
2690                        row("/stop", &out[1]),
2691                        row("/estop", &out[2]),
2692                    ]));
2693                let cell: &'static HazardCell =
2694                    std::boxed::Box::leak(std::boxed::Box::new(HazardCell::new()));
2695                Ladder {
2696                    out,
2697                    rows,
2698                    spec: spec(cell),
2699                }
2700            }
2701            pub fn tick(&self, age: &[AgeState], mode: u16, now: u32) -> std::vec::Vec<(u32, u32)> {
2702                let mon = [MonitorState::default(); 3];
2703                let mut out = std::vec::Vec::new();
2704                check_hazard(&self.spec, self.rows, &mon, age, mode, now, |v| out.push(v));
2705                out.iter()
2706                    .filter(|v| v.rule == "hazard-route-runtime")
2707                    .map(|v| (v.measured, v.declared))
2708                    .collect()
2709            }
2710            pub fn publish(&self, i: usize) {
2711                self.out[i].count.fetch_add(1, Ordering::Relaxed);
2712            }
2713        }
2714        pub fn faulted(silent: [bool; 2]) -> [AgeState; 2] {
2715            let mut age = [AgeState::default(); 2];
2716            for (a, s) in age.iter_mut().zip(silent) {
2717                a.silence_reported = s;
2718            }
2719            age
2720        }
2721    }
2722
2723    /// A fault in the base mode is judged against the TOP rung: its output
2724    /// and its route, from its `starts_at` detection.
2725    #[test]
2726    fn a_fault_at_the_top_is_judged_against_the_top_rung() {
2727        let rig = ladder::Ladder::new();
2728        let age = ladder::faulted([true, false]);
2729        assert!(rig.tick(&age, 0, 1_000).is_empty());
2730        // The floor publishing is not the top rung answering.
2731        rig.publish(2);
2732        assert!(rig.tick(&age, 0, 1_005).is_empty());
2733        assert_eq!(rig.tick(&age, 0, 1_011), [(11, 10)], "top route 10");
2734    }
2735
2736    /// A fault arriving after the ladder has fallen -- the executor is already
2737    /// in the rung's mode -- is judged against THAT rung.
2738    #[test]
2739    fn a_fault_after_a_fall_is_judged_against_the_lower_rung() {
2740        let rig = ladder::Ladder::new();
2741        let age = ladder::faulted([true, false]);
2742        assert!(rig.tick(&age, 2, 2_000).is_empty());
2743        // Only the active rung's output answers it.
2744        rig.publish(0);
2745        assert!(rig.tick(&age, 2, 2_015).is_empty(), "within 20");
2746        assert_eq!(rig.tick(&age, 2, 2_021), [(21, 20)], "rung route 20");
2747        // And it answered within its route on a second fault.
2748        let clear = ladder::faulted([false, false]);
2749        rig.tick(&clear, 2, 2_100);
2750        assert!(rig.tick(&age, 2, 3_000).is_empty());
2751        rig.publish(1);
2752        assert!(rig.tick(&age, 2, 3_019).is_empty(), "answered at 19");
2753        assert!(rig.tick(&age, 2, 3_500).is_empty(), "judged once");
2754    }
2755
2756    /// The window rung answers, its window passes and the ladder falls to the
2757    /// rung below: that rung's clock starts at the FALL (its `starts_at` is
2758    /// the window owner, not a detector here), and it is judged on its own
2759    /// output and route.
2760    #[test]
2761    fn a_fall_from_the_window_rung_starts_the_rung_below_at_the_fall() {
2762        let rig = ladder::Ladder::new();
2763        let age = ladder::faulted([true, false]);
2764        assert!(rig.tick(&age, 0, 1_000).is_empty());
2765        rig.publish(0);
2766        assert!(rig.tick(&age, 1, 1_004).is_empty(), "window rung in time");
2767        // The window passes; the executor enters the rung below at 1_500.
2768        assert!(rig.tick(&age, 2, 1_500).is_empty(), "nothing at the fall");
2769        assert_eq!(
2770            rig.tick(&age, 2, 1_521),
2771            [(21, 20)],
2772            "measured from the fall, not the fire"
2773        );
2774    }
2775
2776    /// The floor: a fall straight to it starts its clock, and a floor that
2777    /// never publishes is reported once when its route passes. A rise back up
2778    /// the ladder is not a fall.
2779    #[test]
2780    fn a_fall_to_the_floor_is_judged_against_the_floor() {
2781        let rig = ladder::Ladder::new();
2782        let age = ladder::faulted([true, false]);
2783        assert!(rig.tick(&age, 0, 1_000).is_empty());
2784        assert_eq!(rig.tick(&age, 0, 1_011), [(11, 10)], "top overdue");
2785        assert!(rig.tick(&age, 3, 1_100).is_empty(), "fall to the floor");
2786        assert!(rig.tick(&age, 1, 1_110).is_empty(), "not a fall");
2787        assert!(rig.tick(&age, 3, 1_130).is_empty(), "30 is not past 30");
2788        assert_eq!(rig.tick(&age, 3, 1_131), [(31, 30)], "floor route 30");
2789        assert!(rig.tick(&age, 3, 1_500).is_empty(), "reported once");
2790    }
2791
2792    /// The top rung's clock starts at ITS `starts_at` detector, not at the
2793    /// first guard member to fault: here member 1 detects first and member 0
2794    /// (the top rung's `starts_at`) only later.
2795    #[test]
2796    fn the_route_clock_starts_at_the_starts_at_detection() {
2797        let rig = ladder::Ladder::new();
2798        assert!(
2799            rig.tick(&ladder::faulted([false, true]), 0, 1_000)
2800                .is_empty()
2801        );
2802        let both = ladder::faulted([true, true]);
2803        assert!(
2804            rig.tick(&both, 0, 1_050).is_empty(),
2805            "50 ms after the fire, but the clock starts now"
2806        );
2807        assert!(rig.tick(&both, 0, 1_060).is_empty(), "10 is not past 10");
2808        assert_eq!(rig.tick(&both, 0, 1_061), [(11, 10)]);
2809    }
2810
2811    /// With no sink the observer never claims a reaction it cannot see.
2812    #[test]
2813    fn no_sink_no_reaction() {
2814        static CELL: HazardCell = HazardCell::new();
2815        static MONITORS: [HazardMonitorSpec; 1] = [HazardMonitorSpec {
2816            all_of: false,
2817            members: &[HazardMonitorRef::Rate(0)],
2818        }];
2819        let spec = HazardSpec {
2820            key: "k",
2821            ftti_ms: 100,
2822            settle_ms: 0,
2823            monitors: &MONITORS,
2824            sink: NO_SINK,
2825            reaction_id: 0,
2826            rungs: &[],
2827            cell: &CELL,
2828        };
2829        let mut mon = [MonitorState::default(); 2];
2830        mon[0].violated_last_window = true;
2831        let v = run(&spec, &mon, &[], 50);
2832        assert_eq!((v[0].rule, v[0].measured), ("hazard-detected", 0));
2833        PUB_CELL.count.fetch_add(5, Ordering::Relaxed);
2834        assert!(run(&spec, &mon, &[], 60).is_empty());
2835    }
2836}
2837
2838#[cfg(test)]
2839mod tests {
2840    use super::*;
2841
2842    static CELL: PubMonitorCell = PubMonitorCell::new();
2843
2844    /// phase-496 W1 -- the Autoware Safety Island's shape: 14 rate rows and
2845    /// no age rows fit an executor sized 14/0, and a 15th rate row is refused
2846    /// with the knob to raise NAMED, not truncated.
2847    #[test]
2848    fn a_table_past_the_knob_is_refused_and_names_it() {
2849        assert_eq!(check_capacity_against(14, 0, 14, 0), Ok(()));
2850        let full = check_capacity_against(15, 0, 14, 0).expect_err("15 > 14 must refuse");
2851        assert_eq!(full.knob, "NROS_EXECUTOR_MAX_MONITORS");
2852        assert_eq!((full.rows, full.capacity), (15, 14));
2853        let said = std::format!("{full}");
2854        assert!(
2855            said.contains("raise NROS_EXECUTOR_MAX_MONITORS"),
2856            "the refusal must name the knob: {said}"
2857        );
2858        let age = check_capacity_against(0, 1, 14, 0).expect_err("an age row past 0 must refuse");
2859        assert_eq!(age.knob, "NROS_EXECUTOR_MAX_AGE_MONITORS");
2860        // The two tables are separate knobs: a full age table does not borrow
2861        // from the rate one.
2862        assert!(check_capacity_against(3, 9, 14, 8).is_err());
2863    }
2864
2865    /// The generated consts ARE the public ones, so the installers and the
2866    /// spin loop read one number.
2867    #[test]
2868    fn the_public_bounds_are_the_generated_config() {
2869        assert_eq!(MAX_MONITORS, crate::config::MAX_MONITORS);
2870        assert_eq!(MAX_AGE_MONITORS, crate::config::MAX_AGE_MONITORS);
2871        assert_eq!(check_table_capacity(MAX_MONITORS, MAX_AGE_MONITORS), Ok(()));
2872        assert!(check_table_capacity(MAX_MONITORS + 1, 0).is_err());
2873    }
2874
2875    fn spec(min_milli: u32) -> MonitorSpec {
2876        MonitorSpec {
2877            topic: "/chatter",
2878            fqn: "/demo/talker/chatter",
2879            min_rate_hz_milli: min_milli,
2880            max_latency_ms: 0,
2881            route_trigger: "",
2882            route_max_latency_ms: 0,
2883            route_deadline_ms: 0,
2884            reaction_id: 0,
2885            mechanism: 0,
2886            cell: &CELL,
2887        }
2888    }
2889
2890    #[test]
2891    fn slow_publisher_fires_once_until_recovery() {
2892        CELL.count.store(0, Ordering::Relaxed);
2893        let s = spec(100_000); // 100 Hz declared
2894        let mut st = MonitorState::default();
2895
2896        // t=0: opens window.
2897        assert!(check_rate(&s, &mut st, 0).is_none());
2898        // 5 publishes in 5 s = 1 Hz — violation.
2899        CELL.count.store(5, Ordering::Relaxed);
2900        let v = check_rate(&s, &mut st, RATE_CHECK_INTERVAL_US).expect("fires");
2901        assert_eq!(v.rule, "rate-hierarchy-runtime");
2902        assert_eq!(v.fqn, "/demo/talker/chatter");
2903        assert_eq!(v.measured, 1_000);
2904        assert_eq!(v.declared, 100_000);
2905        // Still slow next window — suppressed (no re-report spam).
2906        CELL.count.store(10, Ordering::Relaxed);
2907        assert!(check_rate(&s, &mut st, 2 * RATE_CHECK_INTERVAL_US).is_none());
2908        // Recovers (500 publishes in 5 s = 100 Hz) — clean window resets.
2909        CELL.count.store(510, Ordering::Relaxed);
2910        assert!(check_rate(&s, &mut st, 3 * RATE_CHECK_INTERVAL_US).is_none());
2911        // Degrades again — fires again.
2912        CELL.count.store(511, Ordering::Relaxed);
2913        assert!(check_rate(&s, &mut st, 4 * RATE_CHECK_INTERVAL_US).is_some());
2914    }
2915
2916    /// phase-474 I8 -- the island's first window after arming read 10 Hz
2917    /// publishers at 9984 mHz and never again. The monitor tick runs at the
2918    /// top of a spin, before that spin's dispatch publishes, so in steady
2919    /// state every window boundary sits one spin after a publish and the
2920    /// count matches the span. The window arming opens sits wherever the
2921    /// arming spin fell: off the publisher's phase, it spans 50 publishes
2922    /// over 5007 ms and reads short. A stream at exactly its declared rate,
2923    /// armed mid-period, must not be judged slow.
2924    #[test]
2925    fn a_stream_at_its_rate_armed_mid_period_is_not_judged_slow() {
2926        static C8: PubMonitorCell = PubMonitorCell::new();
2927        let s = MonitorSpec {
2928            topic: "/i8",
2929            fqn: "/n/i8",
2930            min_rate_hz_milli: 10_000, // 10 Hz declared
2931            max_latency_ms: 0,
2932            route_trigger: "",
2933            route_max_latency_ms: 0,
2934            route_deadline_ms: 0,
2935            reaction_id: 0,
2936            mechanism: 0,
2937            cell: &C8,
2938        };
2939        // Published at 0, 100 ms, ..., 50 s: 501 samples before arming.
2940        C8.count.store(501, Ordering::Relaxed);
2941        let mut st = MonitorState::default();
2942        // Arming spin, off the 10 ms grid (a wake for another entry).
2943        assert!(check_rate(&s, &mut st, 50_003_000).is_none());
2944        let mut t = 50_010_000u64;
2945        while t <= 80_000_000 {
2946            // The tick, then the spin's dispatch: the 10 Hz timer publishes.
2947            let v = check_rate(&s, &mut st, t);
2948            assert!(v.is_none(), "judged slow at {t} us: {v:?}");
2949            if t.is_multiple_of(100_000) {
2950                C8.count.fetch_add(1, Ordering::Relaxed);
2951            }
2952            t += 10_000;
2953        }
2954        // And a stream that IS slow is still judged in the first window:
2955        // half its rate from arming on.
2956        static C9: PubMonitorCell = PubMonitorCell::new();
2957        let slow = MonitorSpec { cell: &C9, ..s };
2958        let mut st = MonitorState::default();
2959        assert!(check_rate(&slow, &mut st, 50_003_000).is_none());
2960        let mut t = 50_010_000u64;
2961        let mut fired = None;
2962        while t <= 60_000_000 && fired.is_none() {
2963            fired = check_rate(&slow, &mut st, t).map(|v| (t, v.measured));
2964            if t.is_multiple_of(200_000) {
2965                C9.count.fetch_add(1, Ordering::Relaxed);
2966            }
2967            t += 10_000;
2968        }
2969        let (at, measured) = fired.expect("a 5 Hz stream against 10 Hz is judged");
2970        assert!(measured <= 5_100, "{measured}");
2971        // Judged one window after its first sample: at most one of its
2972        // periods (200 ms) plus a tick later than the window arming opened.
2973        assert!(at <= 50_003_000 + RATE_CHECK_INTERVAL_US + 210_000, "{at}");
2974        // A stream silent from arming on is judged at the first window's end,
2975        // not held open waiting for a sample that never comes.
2976        static C10: PubMonitorCell = PubMonitorCell::new();
2977        let dead = MonitorSpec { cell: &C10, ..s };
2978        let mut st = MonitorState::default();
2979        assert!(check_rate(&dead, &mut st, 50_003_000).is_none());
2980        let v = check_rate(&dead, &mut st, 50_003_000 + RATE_CHECK_INTERVAL_US)
2981            .expect("silent stream judged");
2982        assert_eq!(v.measured, 0);
2983    }
2984
2985    /// phase-474 I9 -- the island's 10 Hz publishers read 9999 of 10000 mHz
2986    /// in every window once I8 aligned the first one: the spin that closes a
2987    /// window runs some hundreds of microseconds after the interval, so the
2988    /// quotient floors under the declared rate while the count is exact. The
2989    /// count judges; a stream at its declared rate checked late is not slow,
2990    /// a stream at half its rate still is.
2991    #[test]
2992    fn a_stream_at_its_rate_checked_late_is_not_judged_slow() {
2993        static C11: PubMonitorCell = PubMonitorCell::new();
2994        let s = MonitorSpec {
2995            topic: "/i9",
2996            fqn: "/n/i9",
2997            min_rate_hz_milli: 10_000,
2998            max_latency_ms: 0,
2999            route_trigger: "",
3000            route_max_latency_ms: 0,
3001            route_deadline_ms: 0,
3002            reaction_id: 0,
3003            mechanism: 0,
3004            cell: &C11,
3005        };
3006        let mut st = MonitorState::default();
3007        // Publishes at 0, 100 ms, ...; the monitor tick runs 700 us after
3008        // each publish, so every window spans the interval plus 700 us.
3009        let mut t = 0u64;
3010        while t <= 60_000_000 {
3011            if t.is_multiple_of(100_000) {
3012                C11.count.fetch_add(1, Ordering::Relaxed);
3013            }
3014            let v = check_rate(&s, &mut st, t + 700);
3015            assert!(v.is_none(), "judged slow at {t} us: {v:?}");
3016            t += 10_000;
3017        }
3018        // Half the rate, checked the same way: judged, measured about 5 Hz.
3019        static C12: PubMonitorCell = PubMonitorCell::new();
3020        let slow = MonitorSpec { cell: &C12, ..s };
3021        let mut st = MonitorState::default();
3022        let mut t = 0u64;
3023        let mut fired = None;
3024        while t <= 20_000_000 && fired.is_none() {
3025            if t.is_multiple_of(200_000) {
3026                C12.count.fetch_add(1, Ordering::Relaxed);
3027            }
3028            fired = check_rate(&slow, &mut st, t + 700).map(|v| v.measured);
3029            t += 10_000;
3030        }
3031        let measured = fired.expect("a 5 Hz stream against 10 Hz is judged");
3032        assert!((4_900..=5_100).contains(&measured), "{measured}");
3033    }
3034
3035    #[test]
3036    fn compliant_and_uncontracted_stay_silent() {
3037        static C2: PubMonitorCell = PubMonitorCell::new();
3038        let s = MonitorSpec {
3039            topic: "/t",
3040            fqn: "/n/t",
3041            min_rate_hz_milli: 500, // 0.5 Hz
3042            max_latency_ms: 0,
3043            route_trigger: "",
3044            route_max_latency_ms: 0,
3045            route_deadline_ms: 0,
3046            reaction_id: 0,
3047            mechanism: 0,
3048            cell: &C2,
3049        };
3050        let mut st = MonitorState::default();
3051        assert!(check_rate(&s, &mut st, 0).is_none());
3052        C2.count.store(5, Ordering::Relaxed); // 1 Hz measured ≥ 0.5 Hz declared
3053        assert!(check_rate(&s, &mut st, RATE_CHECK_INTERVAL_US).is_none());
3054
3055        // min_rate 0 = uncontracted: never fires, no state.
3056        let s0 = MonitorSpec {
3057            topic: "/t",
3058            fqn: "/n/t",
3059            min_rate_hz_milli: 0,
3060            max_latency_ms: 0,
3061            route_trigger: "",
3062            route_max_latency_ms: 0,
3063            route_deadline_ms: 0,
3064            reaction_id: 0,
3065            mechanism: 0,
3066            cell: &C2,
3067        };
3068        let mut st0 = MonitorState::default();
3069        assert!(check_rate(&s0, &mut st0, 10 * RATE_CHECK_INTERVAL_US).is_none());
3070    }
3071
3072    #[test]
3073    fn stale_take_fires_age_once_until_recovery() {
3074        static SC: SubMonitorCell = SubMonitorCell::new();
3075        let s = AgeMonitorSpec {
3076            topic: "/scan",
3077            fqn: "/perc/detector/scan",
3078            max_age_ms: 100,
3079            deadline_ms: 0,
3080            cell: &SC,
3081        };
3082        let mut st = AgeState::default();
3083
3084        // Fresh message: stamped 5 ms ago -- silent. Every tick here SEES a
3085        // take, so the silence rule never opens a window (phase-462 W2).
3086        SC.observe(1_000_000_000, 1_000_005_000);
3087        assert!(check_age(&s, &mut st, Some(0)).is_none());
3088        // Stale: 250 ms old — fires with the measured age.
3089        SC.observe(1_000_000_000, 1_000_250_000);
3090        let v = check_age(&s, &mut st, Some(1_000)).expect("fires");
3091        assert_eq!(v.rule, "max-age-runtime");
3092        assert_eq!(v.fqn, "/perc/detector/scan");
3093        assert_eq!(v.measured, 250);
3094        assert_eq!(v.declared, 100);
3095        // Still stale next window — suppressed.
3096        SC.observe(1_000_000_000, 1_000_300_000);
3097        assert!(check_age(&s, &mut st, Some(2_000)).is_none());
3098        // Recovers — clean window resets; stale again refires.
3099        SC.observe(1_000_000_000, 1_000_010_000);
3100        assert!(check_age(&s, &mut st, Some(3_000)).is_none());
3101        SC.observe(1_000_000_000, 1_000_999_000);
3102        assert!(check_age(&s, &mut st, Some(4_000)).is_some());
3103    }
3104
3105    /// phase-462 W2 -- an input that stops coming is a violation of the same
3106    /// `max_age_ms` promise, and the age rule cannot see it.
3107    #[test]
3108    fn a_silent_subscription_fires_once_until_it_is_fed() {
3109        static SC2: SubMonitorCell = SubMonitorCell::new();
3110        let s = AgeMonitorSpec {
3111            topic: "/scan",
3112            fqn: "/perc/detector/scan",
3113            max_age_ms: 100,
3114            deadline_ms: 0,
3115            cell: &SC2,
3116        };
3117        let mut st = AgeState::default();
3118
3119        // t=0 opens the lease; a first check is never a verdict.
3120        SC2.observe(1_000_000_000, 1_000_005_000);
3121        assert!(check_age(&s, &mut st, Some(0)).is_none());
3122        // 50 ms of nothing, inside the declared 100 ms window.
3123        assert!(check_age(&s, &mut st, Some(50_000)).is_none());
3124        // 120 ms of nothing: the lease is up.
3125        let v = check_age(&s, &mut st, Some(120_000)).expect("silence fires");
3126        assert_eq!(v.rule, "silence-runtime");
3127        assert_eq!(v.fqn, "/perc/detector/scan");
3128        assert_eq!(v.measured, 0, "nothing was taken");
3129        assert_eq!(v.declared, 100, "the window the contract declares");
3130        // Still silent -- one fault, not one per tick.
3131        assert!(check_age(&s, &mut st, Some(500_000)).is_none());
3132        // Fed again: the lease restarts and a later silence reports afresh.
3133        SC2.observe(1_000_000_000, 1_000_005_000);
3134        assert!(check_age(&s, &mut st, Some(600_000)).is_none());
3135        assert!(check_age(&s, &mut st, Some(650_000)).is_none());
3136        assert!(check_age(&s, &mut st, Some(800_000)).is_some());
3137    }
3138
3139    /// The negative control for the same rule, in three parts: an
3140    /// uncontracted endpoint, a fed endpoint, and a build with no clock.
3141    #[test]
3142    fn silence_needs_a_contract_a_gap_and_a_clock() {
3143        static SC3: SubMonitorCell = SubMonitorCell::new();
3144        // No age contract: nothing is promised, so nothing is judged.
3145        let uncontracted = AgeMonitorSpec {
3146            topic: "/t",
3147            fqn: "/n/t",
3148            max_age_ms: 0,
3149            deadline_ms: 0,
3150            cell: &SC3,
3151        };
3152        let mut st = AgeState::default();
3153        assert!(check_age(&uncontracted, &mut st, Some(0)).is_none());
3154        assert!(check_age(&uncontracted, &mut st, Some(10_000_000)).is_none());
3155
3156        // Contracted and fed on every tick: silent rule, silent drain.
3157        let s = AgeMonitorSpec {
3158            max_age_ms: 100,
3159            deadline_ms: 0,
3160            ..uncontracted
3161        };
3162        let mut st = AgeState::default();
3163        for tick in 0..10u64 {
3164            SC3.observe(1_000_000_000, 1_000_001_000);
3165            assert!(
3166                check_age(&s, &mut st, Some(tick * 1_000_000)).is_none(),
3167                "a fed endpoint never reports silence"
3168            );
3169        }
3170
3171        // No clock: degrade, never guess. A window measured in spins would be
3172        // a tolerance nobody declared.
3173        let mut st = AgeState::default();
3174        assert!(check_age(&s, &mut st, None).is_none());
3175        assert!(check_age(&s, &mut st, None).is_none());
3176    }
3177
3178    #[test]
3179    fn peek_stamp_reads_le_time_and_rejects_unstamped() {
3180        // Encapsulation header (4B) + sec=100 nsec=5000 at offset 4.
3181        let mut raw = [0u8; 12];
3182        raw[4..8].copy_from_slice(&100i32.to_le_bytes());
3183        raw[8..12].copy_from_slice(&5_000u32.to_le_bytes());
3184        assert_eq!(peek_stamp_us(&raw, 4), Some(100_000_005));
3185        // Zero / negative sec = unstamped: no age sample.
3186        assert_eq!(peek_stamp_us(&[0u8; 12], 4), None);
3187        // Short buffer: no panic, no sample.
3188        assert_eq!(peek_stamp_us(&raw[..8], 4), None);
3189    }
3190
3191    #[test]
3192    fn slow_path_fires_latency_once_until_recovery() {
3193        static C3: PubMonitorCell = PubMonitorCell::new();
3194        let s = MonitorSpec {
3195            topic: "/cmd",
3196            fqn: "/ctrl/control/cmd",
3197            min_rate_hz_milli: 0,
3198            max_latency_ms: 10,
3199            route_trigger: "",
3200            route_max_latency_ms: 0,
3201            route_deadline_ms: 0,
3202            reaction_id: 0,
3203            mechanism: 0,
3204            cell: &C3,
3205        };
3206        let mut st = MonitorState::default();
3207        // 4 ms dispatch — within budget.
3208        C3.max_latency_us.store(4_000, Ordering::Relaxed);
3209        assert!(check_latency(&s, &mut st).is_none());
3210        assert_eq!(C3.max_latency_us.load(Ordering::Relaxed), 0, "drained");
3211        // 25 ms dispatch — fires.
3212        C3.max_latency_us.store(25_000, Ordering::Relaxed);
3213        let v = check_latency(&s, &mut st).expect("fires");
3214        assert_eq!(v.rule, "max-latency-runtime");
3215        assert_eq!(v.measured, 25);
3216        assert_eq!(v.declared, 10);
3217        // Still slow — suppressed; recovery resets.
3218        C3.max_latency_us.store(30_000, Ordering::Relaxed);
3219        assert!(check_latency(&s, &mut st).is_none());
3220        C3.max_latency_us.store(1_000, Ordering::Relaxed);
3221        assert!(check_latency(&s, &mut st).is_none());
3222        C3.max_latency_us.store(30_000, Ordering::Relaxed);
3223        assert!(check_latency(&s, &mut st).is_some());
3224    }
3225}
3226
3227/// Issue #514 — emit one violation to the log.
3228///
3229/// A log line is deliberately the floor rather than a `/diagnostics`
3230/// publication: it needs no publisher, no topic wiring, and no contract
3231/// on the reporting path itself, so it works on a bare RTOS image and
3232/// during boot. Publishing the same verdicts as `DiagnosticArray`
3233/// belongs on top of this, not instead of it.
3234///
3235/// Free function rather than an `Executor` method because every call
3236/// site sits inside a loop that already borrows the executor's spec
3237/// tables.
3238pub(crate) fn log_violation(v: &Violation) {
3239    nros_log::log_warn!(
3240        nros_log::get_logger("nros"),
3241        "contract violation: {} {} measured={} declared={}",
3242        v.rule,
3243        v.fqn,
3244        v.measured,
3245        v.declared
3246    );
3247}
3248
3249/// phase-474 I1 -- the drain-and-report hook's line: the issue-0514 line plus
3250/// the verdict's sequence number and the counters, so a reader who sees line
3251/// `#12` after `#9` knows two were missed, and `dropped` says whether the ring
3252/// lost any before this drain.
3253fn log_drained_violation(seq: u32, total: u32, dropped: u32, v: &Violation) {
3254    nros_log::log_warn!(
3255        nros_log::get_logger("nros"),
3256        "contract violation #{}: {} {} measured={} declared={} (total={} dropped={})",
3257        seq,
3258        v.rule,
3259        v.fqn,
3260        v.measured,
3261        v.declared,
3262        total,
3263        dropped
3264    );
3265}
3266
3267/// Issue #505 — periodic activations a timer dropped because its
3268/// executor was blocked past the period boundary.
3269///
3270/// This rule exists because `check_rate` cannot see an isolated stall:
3271/// it samples publish counts over a ~5 s window, so 20 missed
3272/// activations of a 100 Hz loop are a 0.4% rate deficit — under any
3273/// sane declared minimum, silence. (And under
3274/// [`TimerOverrunPolicy::CatchUp`](super::arena::TimerOverrunPolicy)
3275/// the replayed activations refill the window entirely, so the rate
3276/// rule reports a HEALTHY loop while the tier is stalling.) The
3277/// overrun counter is exact, needs no window, and does not depend on
3278/// clock resolution.
3279///
3280/// `overruns` is the timer's monotonic saturating counter;
3281/// `last_reported` is the value at the previous check, so the verdict
3282/// is on the delta. Returns a violation when more than `tolerated`
3283/// activations were dropped since the last check.
3284pub(crate) fn check_timer_overrun(
3285    overruns: u32,
3286    last_reported: &mut u32,
3287    tolerated: u32,
3288) -> Option<Violation> {
3289    let dropped = overruns.saturating_sub(*last_reported);
3290    *last_reported = overruns;
3291    if dropped <= tolerated {
3292        return None;
3293    }
3294    Some(Violation {
3295        rule: "timer-overrun-runtime",
3296        // Timer entries carry no name at this altitude; same stand-in
3297        // as `deadline-miss-runtime`.
3298        fqn: "timer",
3299        measured: dropped,
3300        declared: tolerated,
3301    })
3302}
3303
3304/// Issue #515 — report a spin wake that arrived so late the cadence it
3305/// claims cannot have been met.
3306///
3307/// Like `check_timer_overrun` and unlike the rate/age/latency rules, this
3308/// needs no baked spec table. The bound is the spin period ITSELF: the
3309/// caller passes it to `spin_once` as the pacing quantum, it is what
3310/// `system.toml` declares as `spin_period_us`, and a wake later than a full
3311/// period past its predecessor means an activation's worth of cadence was
3312/// lost. That is a contract failure for any declared period, so there is
3313/// nothing further to declare.
3314///
3315/// The tolerance is one whole period rather than zero, and deliberately so.
3316/// Sub-period lateness is ordinary scheduling noise -- on the measured FVP
3317/// lane the executor is late on a large fraction of wakes while still
3318/// holding its rate -- and a rule that fired on each one would report a
3319/// healthy system as broken thousands of times a second. What is NOT
3320/// ordinary is being a full period late, because by then the wake that
3321/// should have happened in between never did.
3322///
3323/// phase-474 I11 -- for a `spin_once(period)` loop with no declared cadence
3324/// the lateness is release minus scheduled release on the executor's release
3325/// grid (a dropped release counts its age when dropped), not entry-to-entry
3326/// minus the period; see `Executor::record_release_jitter`.
3327///
3328/// `max_jitter_us` is the executor's high-water since the last check and
3329/// `last_reported` the value at the previous one, so the verdict is on the
3330/// delta -- the same shape as the overrun counter, and for the same reason:
3331/// a maximum that has not moved is not a new fault.
3332pub(crate) fn check_release_jitter(
3333    max_jitter_us: u64,
3334    last_reported: &mut u64,
3335    period_us: u64,
3336) -> Option<Violation> {
3337    if period_us == 0 || max_jitter_us <= *last_reported {
3338        return None;
3339    }
3340    *last_reported = max_jitter_us;
3341    if max_jitter_us < period_us {
3342        return None;
3343    }
3344    Some(Violation {
3345        rule: "release-jitter-runtime",
3346        // Same stand-in as the timer and deadline rules: the spin loop is
3347        // not an endpoint and carries no fqn at this altitude.
3348        fqn: "spin",
3349        measured: max_jitter_us.min(u32::MAX as u64) as u32,
3350        declared: period_us.min(u32::MAX as u64) as u32,
3351    })
3352}
3353
3354/// Report a spin thread whose stack has come closer to its end than the
3355/// declared minimum.
3356///
3357/// Unlike every other rule here the bound CANNOT be derived from something
3358/// already declared, and that is worth stating rather than papering over.
3359/// `check_timer_overrun` and `check_release_jitter` both judge against a
3360/// period the caller already passes in; there is no equivalent for a stack.
3361/// The executor never sees `stack_bytes` -- it lives in the spawn attr and
3362/// goes no further -- and the total is not portably queryable either:
3363/// FreeRTOS exposes the high-water mark and not the size it was taken
3364/// against, so even a percentage cannot be computed. A minimum headroom is
3365/// therefore a real declaration, and `min_bytes == 0` means the caller has
3366/// not made one, which disables the rule.
3367///
3368/// Reports on the WORST case, not on each crossing: `worst_reported` holds
3369/// the lowest headroom already reported, so a stack hovering just under the
3370/// bound says so once and then only when it gets worse. The same delta
3371/// discipline as the overrun and jitter rules, inverted because for headroom
3372/// smaller is worse.
3373pub(crate) fn check_stack_headroom(
3374    unused_bytes: usize,
3375    min_bytes: usize,
3376    worst_reported: &mut usize,
3377) -> Option<Violation> {
3378    if min_bytes == 0 || unused_bytes >= min_bytes {
3379        return None;
3380    }
3381    // `usize::MAX` is the "nothing reported yet" sentinel: any real headroom
3382    // is below it, so the first breach always reports.
3383    if *worst_reported != usize::MAX && unused_bytes >= *worst_reported {
3384        return None;
3385    }
3386    *worst_reported = unused_bytes;
3387    Some(Violation {
3388        rule: "stack-headroom-runtime",
3389        // The spin thread is not an endpoint; same stand-in as the timer,
3390        // deadline and jitter rules.
3391        fqn: "stack",
3392        measured: unused_bytes.min(u32::MAX as usize) as u32,
3393        declared: min_bytes.min(u32::MAX as usize) as u32,
3394    })
3395}
3396
3397#[cfg(test)]
3398mod stack_headroom_rule_tests {
3399    use super::*;
3400
3401    /// No declared minimum means no claim to breach.
3402    #[test]
3403    fn a_zero_minimum_disables_the_rule() {
3404        let mut worst = usize::MAX;
3405        assert!(check_stack_headroom(8, 0, &mut worst).is_none());
3406    }
3407
3408    #[test]
3409    fn headroom_at_the_bound_is_not_a_breach() {
3410        let mut worst = usize::MAX;
3411        assert!(check_stack_headroom(1024, 1024, &mut worst).is_none());
3412        assert!(check_stack_headroom(2048, 1024, &mut worst).is_none());
3413    }
3414
3415    #[test]
3416    fn reports_the_first_breach_with_both_numbers() {
3417        let mut worst = usize::MAX;
3418        let v = check_stack_headroom(512, 1024, &mut worst).expect("under the bound reports");
3419        assert_eq!(v.rule, "stack-headroom-runtime");
3420        assert_eq!(v.measured, 512);
3421        assert_eq!(v.declared, 1024);
3422    }
3423
3424    /// Smaller is worse for headroom, so the delta runs the other way.
3425    #[test]
3426    fn only_a_new_low_is_a_new_fault() {
3427        let mut worst = usize::MAX;
3428        assert!(check_stack_headroom(512, 1024, &mut worst).is_some());
3429        assert!(check_stack_headroom(512, 1024, &mut worst).is_none());
3430        assert!(check_stack_headroom(600, 1024, &mut worst).is_none());
3431        let v = check_stack_headroom(100, 1024, &mut worst).expect("a new low reports");
3432        assert_eq!(v.measured, 100);
3433    }
3434}
3435
3436#[cfg(test)]
3437mod publish_stamp_tests {
3438    use super::*;
3439
3440    /// CDR: 4-byte encapsulation header, then `Time { i32 sec; u32 nanosec }`
3441    /// little-endian, so `sec` sits at byte 4 — the layout `STAMP_OFFSET`
3442    /// encodes.
3443    fn cdr_with_stamp(sec: i32, nanosec: u32) -> [u8; 12] {
3444        let mut b = [0u8; 12];
3445        b[4..8].copy_from_slice(&sec.to_le_bytes());
3446        b[8..12].copy_from_slice(&nanosec.to_le_bytes());
3447        b
3448    }
3449
3450    #[test]
3451    fn records_the_age_of_what_was_published() {
3452        let cell = PubMonitorCell::new();
3453        let raw = cdr_with_stamp(10, 0); // stamped at 10_000_000 us
3454        observe_publish_stamp(&cell, &raw, 4, 10_500_000);
3455        assert_eq!(
3456            cell.last_publish_stamp_age_us.load(Ordering::Relaxed),
3457            500_000,
3458            "published data was half a second old"
3459        );
3460    }
3461
3462    /// The signature of a node that RE-STAMPED: it publishes data whose
3463    /// stamp is now, so downstream age is single-hop, not end-to-end.
3464    #[test]
3465    fn a_restamping_node_shows_near_zero_age() {
3466        let cell = PubMonitorCell::new();
3467        let raw = cdr_with_stamp(10, 0);
3468        observe_publish_stamp(&cell, &raw, 4, 10_000_000);
3469        assert_eq!(cell.last_publish_stamp_age_us.load(Ordering::Relaxed), 0);
3470    }
3471
3472    /// A state, not a window maximum: one stale message at startup must not
3473    /// pin the value for the life of the process.
3474    #[test]
3475    fn the_latest_publish_replaces_the_previous() {
3476        let cell = PubMonitorCell::new();
3477        observe_publish_stamp(&cell, &cdr_with_stamp(10, 0), 4, 12_000_000);
3478        assert_eq!(
3479            cell.last_publish_stamp_age_us.load(Ordering::Relaxed),
3480            2_000_000
3481        );
3482        observe_publish_stamp(&cell, &cdr_with_stamp(20, 0), 4, 20_100_000);
3483        assert_eq!(
3484            cell.last_publish_stamp_age_us.load(Ordering::Relaxed),
3485            100_000,
3486            "a fresh publish replaces the old age rather than maxing with it"
3487        );
3488    }
3489
3490    /// An unset stamp is not an age of `now`. `peek_stamp_us` rejects
3491    /// `sec <= 0`, so a zeroed header records nothing at all.
3492    #[test]
3493    fn an_unstamped_message_records_nothing() {
3494        let cell = PubMonitorCell::new();
3495        observe_publish_stamp(&cell, &cdr_with_stamp(0, 0), 4, 5_000_000);
3496        assert_eq!(
3497            cell.last_publish_stamp_age_us.load(Ordering::Relaxed),
3498            0,
3499            "no stamp means no observation, not an enormous age"
3500        );
3501    }
3502}
3503
3504/// Per-SchedContext liveness accounting, one per SC slot.
3505///
3506/// `Default` is DERIVED, not hand-written: every field's default is its type's
3507/// own (`false`, `0`), so a manual impl restates four of them and is one edit
3508/// away from disagreeing with the struct — `clippy::derivable_impls` is a
3509/// `-D warnings` error in the embedded lane, which is where this was caught.
3510#[derive(Debug, Clone, Copy, Default)]
3511pub(crate) struct AliveState {
3512    pub opened: bool,
3513    pub window_start_us: u64,
3514    pub count_at_window_start: u32,
3515    /// Report-once-until-recovery, matching the rate rule.
3516    pub violated_last_window: bool,
3517}
3518
3519/// One SchedContext's alive-supervision record: the counter the dispatch
3520/// path bumps and the window state the monitor tick reads.
3521///
3522/// A region of the carved backing (`storage::carve`), one per SC the
3523/// executor was sized for — NOT an inline `[_; MAX_SC]` in the header.
3524/// `MAX_SC` is a sizing knob, so an inline table grows the `Executor` value
3525/// with it, which is the regression issue 0961 exists to prevent.
3526#[derive(Debug, Clone, Copy, Default)]
3527pub(crate) struct AliveSlot {
3528    pub dispatches: u32,
3529    pub state: AliveState,
3530}
3531
3532/// Report a SchedContext that has not dispatched AT ALL over a full window
3533/// while declaring a period that says it should have.
3534///
3535/// AUTOSAR's Watchdog Manager separates ALIVE supervision -- did it run at
3536/// all, at roughly the right rate -- from DEADLINE supervision, did it finish
3537/// in time. Every rule here is the second kind or a variant of it: they fire
3538/// when something happens and is wrong. Nothing fires when a callback stops
3539/// happening altogether.
3540///
3541/// `rate-hierarchy-runtime` does not cover it. That rule counts PUBLISHES on
3542/// a contracted publisher, so a timer callback that silently stops firing
3543/// while its topic is published from elsewhere -- another tier, a bridge, a
3544/// republisher -- leaves it satisfied. The gap is a callback, not an
3545/// endpoint.
3546///
3547/// The bound needs no declaration: `period_us` is already on the SC, and a
3548/// context claiming a period that produced no activation in a window many
3549/// times longer than it is a contract failure for any period. `period_us ==
3550/// 0` declares no cadence, so nothing is judged.
3551///
3552/// ZERO, not a fraction. A throttled-but-alive context is a different fault
3553/// with different evidence -- `SporadicState` already counts window skips and
3554/// dispatches for exactly that -- and picking a percentage here would invent
3555/// a tolerance nobody declared. Zero activations against a declared period is
3556/// unambiguous.
3557pub(crate) fn check_alive(
3558    period_us: u64,
3559    dispatches: u32,
3560    state: &mut AliveState,
3561    now_us: u64,
3562) -> Option<Violation> {
3563    if period_us == 0 {
3564        return None;
3565    }
3566    if !state.opened {
3567        state.opened = true;
3568        state.window_start_us = now_us;
3569        state.count_at_window_start = dispatches;
3570        return None;
3571    }
3572    let window_us = now_us.saturating_sub(state.window_start_us);
3573    if window_us < RATE_CHECK_INTERVAL_US {
3574        return None;
3575    }
3576    let fired = dispatches.wrapping_sub(state.count_at_window_start);
3577    state.window_start_us = now_us;
3578    state.count_at_window_start = dispatches;
3579
3580    // Only judge a window long enough to have contained an activation. A
3581    // period longer than the check interval is not late merely for not having
3582    // fired yet.
3583    if window_us < period_us {
3584        return None;
3585    }
3586    if fired > 0 {
3587        state.violated_last_window = false;
3588        return None;
3589    }
3590    if state.violated_last_window {
3591        return None; // still silent — already reported
3592    }
3593    state.violated_last_window = true;
3594    Some(Violation {
3595        rule: "alive-supervision-runtime",
3596        // SCs carry no name at this altitude; same stand-in as the deadline
3597        // rule, which reports against the SC too.
3598        fqn: "sched-context",
3599        measured: 0,
3600        declared: (window_us / period_us).min(u32::MAX as u64) as u32,
3601    })
3602}
3603
3604#[cfg(test)]
3605mod alive_supervision_tests {
3606    use super::*;
3607
3608    const W: u64 = RATE_CHECK_INTERVAL_US;
3609
3610    #[test]
3611    fn a_zero_period_declares_no_cadence() {
3612        let mut st = AliveState::default();
3613        assert!(check_alive(0, 0, &mut st, 0).is_none());
3614        assert!(check_alive(0, 0, &mut st, W * 4).is_none());
3615    }
3616
3617    /// The first observation opens the window; there is no verdict yet.
3618    #[test]
3619    fn the_first_window_only_opens() {
3620        let mut st = AliveState::default();
3621        assert!(check_alive(1_000, 0, &mut st, 0).is_none());
3622        assert!(st.opened);
3623    }
3624
3625    #[test]
3626    fn silence_across_a_full_window_reports() {
3627        let mut st = AliveState::default();
3628        check_alive(1_000, 0, &mut st, 0);
3629        let v = check_alive(1_000, 0, &mut st, W).expect("no dispatch in a full window");
3630        assert_eq!(v.rule, "alive-supervision-runtime");
3631        assert_eq!(v.measured, 0);
3632        assert_eq!(v.declared, (W / 1_000) as u32);
3633    }
3634
3635    /// Any activation at all clears it -- this rule asks whether the context
3636    /// is alive, not whether it is keeping up.
3637    #[test]
3638    fn a_single_dispatch_is_enough() {
3639        let mut st = AliveState::default();
3640        check_alive(1_000, 0, &mut st, 0);
3641        assert!(check_alive(1_000, 1, &mut st, W).is_none());
3642    }
3643
3644    /// Report once, then stay quiet until it recovers, like the rate rule.
3645    #[test]
3646    fn continued_silence_is_not_a_new_fault() {
3647        let mut st = AliveState::default();
3648        check_alive(1_000, 0, &mut st, 0);
3649        assert!(check_alive(1_000, 0, &mut st, W).is_some());
3650        assert!(check_alive(1_000, 0, &mut st, W * 2).is_none());
3651        // Recovery, then silence again, reports afresh.
3652        assert!(check_alive(1_000, 5, &mut st, W * 3).is_none());
3653        assert!(check_alive(1_000, 5, &mut st, W * 4).is_some());
3654    }
3655
3656    /// A period longer than the window has not had its chance yet.
3657    #[test]
3658    fn a_period_longer_than_the_window_is_not_judged() {
3659        let mut st = AliveState::default();
3660        check_alive(W * 10, 0, &mut st, 0);
3661        assert!(check_alive(W * 10, 0, &mut st, W).is_none());
3662    }
3663}
3664
3665#[cfg(test)]
3666mod release_jitter_rule_tests {
3667    use super::*;
3668
3669    /// Sub-period lateness is noise, not a violation -- otherwise a
3670    /// healthy-but-jittery loop reports thousands of faults a second.
3671    #[test]
3672    fn tolerates_lateness_within_one_period() {
3673        let mut last = 0;
3674        assert!(check_release_jitter(4_000, &mut last, 5_000).is_none());
3675        assert_eq!(last, 4_000, "still recorded, so the next delta is honest");
3676    }
3677
3678    /// A full period late means the wake that belonged in between never
3679    /// happened.
3680    #[test]
3681    fn reports_a_wake_a_whole_period_late() {
3682        let mut last = 0;
3683        let v = check_release_jitter(5_000, &mut last, 5_000).expect("one period late reports");
3684        assert_eq!(v.rule, "release-jitter-runtime");
3685        assert_eq!(v.measured, 5_000);
3686        assert_eq!(v.declared, 5_000);
3687    }
3688
3689    /// The verdict is on the DELTA. A high-water that has not moved is the
3690    /// same fault already reported, not a new one.
3691    #[test]
3692    fn an_unchanged_maximum_is_not_a_new_fault() {
3693        let mut last = 0;
3694        assert!(check_release_jitter(9_000, &mut last, 5_000).is_some());
3695        assert!(check_release_jitter(9_000, &mut last, 5_000).is_none());
3696        assert!(check_release_jitter(12_000, &mut last, 5_000).is_some());
3697    }
3698
3699    /// No declared period means no cadence to be late for.
3700    #[test]
3701    fn a_zero_period_declares_nothing() {
3702        let mut last = 0;
3703        assert!(check_release_jitter(1_000_000, &mut last, 0).is_none());
3704    }
3705}
3706
3707#[cfg(test)]
3708mod timer_overrun_rule_tests {
3709    use super::*;
3710
3711    #[test]
3712    fn reports_the_delta_not_the_total() {
3713        let mut last = 0;
3714        let v = check_timer_overrun(19, &mut last, 0).expect("first drop reports");
3715        assert_eq!(v.rule, "timer-overrun-runtime");
3716        assert_eq!(v.measured, 19);
3717        // Same total on the next check is not a new fault.
3718        assert!(check_timer_overrun(19, &mut last, 0).is_none());
3719        // Only the newly dropped activations are reported.
3720        assert_eq!(check_timer_overrun(25, &mut last, 0).unwrap().measured, 6);
3721    }
3722
3723    #[test]
3724    fn a_clean_timer_never_reports() {
3725        let mut last = 0;
3726        for _ in 0..10 {
3727            assert!(check_timer_overrun(0, &mut last, 0).is_none());
3728        }
3729    }
3730
3731    #[test]
3732    fn tolerance_suppresses_small_drops() {
3733        let mut last = 0;
3734        assert!(check_timer_overrun(2, &mut last, 2).is_none());
3735        assert_eq!(check_timer_overrun(6, &mut last, 2).unwrap().measured, 4);
3736    }
3737}
3738
3739// phase-436 E4 — when the stack-headroom query is due. The decision is pure so
3740// the three cases are pinned without a clock or a port.
3741#[cfg(test)]
3742mod stack_headroom_throttle_tests {
3743    use super::*;
3744
3745    #[test]
3746    fn the_first_check_is_immediate_with_or_without_a_clock() {
3747        assert!(stack_headroom_check_due(Some(5), None, 0));
3748        assert!(stack_headroom_check_due(None, None, 0));
3749    }
3750
3751    #[test]
3752    fn a_clocked_check_waits_one_full_interval() {
3753        let i = STACK_HEADROOM_CHECK_INTERVAL_US;
3754        assert!(!stack_headroom_check_due(
3755            Some(1_000 + i - 1),
3756            Some(1_000),
3757            0
3758        ));
3759        assert!(stack_headroom_check_due(Some(1_000 + i), Some(1_000), 0));
3760    }
3761
3762    #[test]
3763    fn a_clockless_build_checks_once_per_stride_of_spins() {
3764        let s = STACK_HEADROOM_CHECK_SPIN_STRIDE;
3765        assert!(!stack_headroom_check_due(None, Some(0), s - 1));
3766        assert!(stack_headroom_check_due(None, Some(0), s));
3767    }
3768}
3769
3770/// phase-474 I1 -- the wire vocabulary and the SWD record.
3771#[cfg(test)]
3772mod ring_record_tests {
3773    use super::*;
3774
3775    fn v(rule: &'static str, fqn: &'static str, measured: u32) -> Violation {
3776        Violation {
3777            rule,
3778            fqn,
3779            measured,
3780            declared: 10,
3781        }
3782    }
3783
3784    #[test]
3785    fn every_rule_the_executor_reports_has_a_stable_code() {
3786        // The order is the wire format: a capture or a RAM dump decodes by it.
3787        assert_eq!(rule_code("rate-hierarchy-runtime"), 1);
3788        assert_eq!(rule_code("max-latency-runtime"), 3);
3789        assert_eq!(rule_code("release-jitter-runtime"), 9);
3790        assert_eq!(rule_code("no-such-rule"), 0);
3791        for (i, r) in RULE_IDS.iter().enumerate() {
3792            assert_eq!(rule_name(rule_code(r)), Some(*r), "code {}", i + 1);
3793        }
3794        assert_eq!(rule_name(0), None);
3795        assert_eq!(rule_name(RULE_IDS.len() as u32 + 1), None);
3796    }
3797
3798    #[test]
3799    fn fqn_hash_is_fnv1a_32() {
3800        // Reference values of 32-bit FNV-1a, so a Python decoder can match.
3801        assert_eq!(fqn_hash(""), 0x811c_9dc5);
3802        assert_eq!(fqn_hash("a"), 0xe40c_292c);
3803        assert_eq!(fqn_hash("foobar"), 0xbf9c_f968);
3804    }
3805
3806    #[test]
3807    fn a_violation_is_four_marker_events() {
3808        let w = violation_marker_words(5, &v("max-latency-runtime", "/n/out", 25));
3809        assert_eq!(w[0], (MARKER_VIOLATION, (5 << 8) | 3));
3810        assert_eq!(w[1], (MARKER_VIOLATION_FQN, fqn_hash("/n/out")));
3811        assert_eq!(w[2], (MARKER_VIOLATION_MEASURED, 25));
3812        assert_eq!(w[3], (MARKER_VIOLATION_DECLARED, 10));
3813        // The sequence number keeps its low 24 bits; the rule its low 8.
3814        let w = violation_marker_words(0x0100_0007, &v("silence-runtime", "/x", 0));
3815        assert_eq!(w[0].1, (7 << 8) | 7);
3816    }
3817
3818    #[test]
3819    fn the_record_keeps_the_latest_and_counts_what_it_overwrote() {
3820        let r: ViolationRecord<3> = ViolationRecord::new();
3821        assert_eq!(r.magic.load(Ordering::Relaxed), RECORD_MAGIC);
3822        assert_eq!(r.capacity.load(Ordering::Relaxed), 3);
3823        for m in 1..=5 {
3824            assert_eq!(r.store(&v("timer-overrun-runtime", "timer", m)), m);
3825        }
3826        assert_eq!(r.total.load(Ordering::Relaxed), 5);
3827        assert_eq!(r.dropped.load(Ordering::Relaxed), 2);
3828        assert_eq!(r.head.load(Ordering::Relaxed), 5 % 3);
3829        let mut seen = std::vec::Vec::new();
3830        r.newest_first(|seq, rule, fqn, measured, declared| {
3831            assert_eq!(rule, rule_code("timer-overrun-runtime"));
3832            assert_eq!(fqn, fqn_hash("timer"));
3833            assert_eq!(declared, 10);
3834            seen.push((seq, measured));
3835        });
3836        assert_eq!(seen, [(5, 5), (4, 4), (3, 3)], "newest first, oldest gone");
3837    }
3838
3839    /// The layout a debugger decodes: `u32` words only, header then slots, the
3840    /// same size on every target. `scripts/read-violation-record.py` reads it
3841    /// by these numbers.
3842    #[test]
3843    fn the_record_is_packed_u32_words() {
3844        use core::mem::{offset_of, size_of};
3845        assert_eq!(size_of::<ViolationSlot>(), RECORD_SLOT_WORDS as usize * 4);
3846        assert_eq!(
3847            offset_of!(ViolationRecord<4>, slots),
3848            RECORD_HEADER_WORDS as usize * 4
3849        );
3850        assert_eq!(
3851            size_of::<ViolationRecord<4>>(),
3852            (RECORD_HEADER_WORDS + 4 * RECORD_SLOT_WORDS) as usize * 4
3853        );
3854        assert_eq!(offset_of!(ViolationRecord<4>, total), 16);
3855        assert_eq!(offset_of!(ViolationRecord<4>, head), 20);
3856        assert_eq!(offset_of!(ViolationRecord<4>, dropped), 24);
3857        assert_eq!(offset_of!(ViolationRecord<4>, suppressed_before_arm), 28);
3858        assert_eq!(offset_of!(ViolationRecord<4>, armed), 32);
3859        assert_eq!(offset_of!(ViolationSlot, fqn_addr), 20);
3860    }
3861}
3862
3863/// phase10-B3 -- the route rule's pure half: binding, the earliest unanswered
3864/// take, judging only on a publish of the output, report once until clean.
3865#[cfg(test)]
3866mod route_tests {
3867    use super::*;
3868
3869    static CELL: PubMonitorCell = PubMonitorCell::new();
3870    static OTHER: PubMonitorCell = PubMonitorCell::new();
3871    static TABLE: [MonitorSpec; 2] = [
3872        MonitorSpec {
3873            topic: "/out",
3874            fqn: "/n/out",
3875            min_rate_hz_milli: 0,
3876            max_latency_ms: 0,
3877            route_trigger: "/in",
3878            route_max_latency_ms: 149,
3879            route_deadline_ms: 0,
3880            reaction_id: 0,
3881            mechanism: 0,
3882            cell: &CELL,
3883        },
3884        MonitorSpec {
3885            topic: "/other",
3886            fqn: "/n/other",
3887            min_rate_hz_milli: 0,
3888            max_latency_ms: 0,
3889            route_trigger: "",
3890            route_max_latency_ms: 0,
3891            route_deadline_ms: 0,
3892            reaction_id: 0,
3893            mechanism: 0,
3894            cell: &OTHER,
3895        },
3896    ];
3897
3898    fn snapshot() -> [u32; MAX_MONITORS] {
3899        let mut c = [0u32; MAX_MONITORS];
3900        for (k, s) in TABLE.iter().enumerate() {
3901            c[k] = s.cell.count.load(Ordering::Relaxed);
3902        }
3903        c
3904    }
3905
3906    fn judge(end_us: u64, before: &[u32; MAX_MONITORS]) -> Vec<(u32, u32)> {
3907        let mut out = Vec::new();
3908        route_judge(&TABLE, before, end_us, |_, v, record, react| {
3909            assert!(record && !react, "an unsupervised route only records");
3910            assert_eq!(v.rule, "route-latency-runtime");
3911            assert_eq!(v.fqn, "/n/out");
3912            out.push((v.measured, v.declared));
3913        });
3914        out
3915    }
3916
3917    extern crate alloc;
3918    use alloc::vec::Vec;
3919
3920    #[test]
3921    fn route_rule() {
3922        assert!(route_active(&TABLE));
3923        route_bind(&TABLE, "/elsewhere", 1);
3924        assert_eq!(
3925            CELL.route.load(Ordering::Relaxed),
3926            0,
3927            "wrong topic binds nothing"
3928        );
3929        route_bind(&TABLE, "/in", 3);
3930        route_bind(&TABLE, "/in", 5);
3931        assert_eq!(
3932            CELL.route.load(Ordering::Relaxed),
3933            4,
3934            "first subscription keeps it"
3935        );
3936        assert_eq!(OTHER.route.load(Ordering::Relaxed), 0);
3937
3938        // A take on another slot starts nothing; two takes keep the earliest.
3939        route_take(&TABLE, 5, 1_000);
3940        assert_eq!(CELL.route.load(Ordering::Relaxed) & ROUTE_PENDING, 0);
3941        route_take(&TABLE, 3, 10_000);
3942        route_take(&TABLE, 3, 90_000);
3943        // A dispatch that publishes nothing judges nothing.
3944        let b = snapshot();
3945        assert!(judge(400_000, &b).is_empty());
3946        assert_ne!(CELL.route.load(Ordering::Relaxed) & ROUTE_PENDING, 0);
3947        // The output published: 10 ms -> 160 ms is 150 ms, over 149.
3948        let b = snapshot();
3949        CELL.count.fetch_add(1, Ordering::Relaxed);
3950        assert_eq!(judge(160_000, &b), [(150, 149)]);
3951        assert_eq!(
3952            CELL.route.load(Ordering::Relaxed) & ROUTE_PENDING,
3953            0,
3954            "answered"
3955        );
3956        // Over again: reported once until a route is within the budget.
3957        route_take(&TABLE, 3, 200_000);
3958        let b = snapshot();
3959        CELL.count.fetch_add(1, Ordering::Relaxed);
3960        assert!(
3961            judge(400_000, &b).is_empty(),
3962            "still over: not reported twice"
3963        );
3964        route_take(&TABLE, 3, 500_000);
3965        let b = snapshot();
3966        CELL.count.fetch_add(1, Ordering::Relaxed);
3967        assert!(judge(600_000, &b).is_empty(), "100 ms: within");
3968        assert_eq!(CELL.route.load(Ordering::Relaxed) & ROUTE_VIOLATED, 0);
3969        route_take(&TABLE, 3, 700_000);
3970        let b = snapshot();
3971        CELL.count.fetch_add(1, Ordering::Relaxed);
3972        assert_eq!(
3973            judge(1_000_000, &b),
3974            [(300, 149)],
3975            "over after clean: reported"
3976        );
3977        // A publish with no waiting take is not a route.
3978        let b = snapshot();
3979        CELL.count.fetch_add(1, Ordering::Relaxed);
3980        assert!(judge(9_000_000, &b).is_empty());
3981        // Arming drops a waiting take and the report-once state, keeps the
3982        // binding; a release unbinds.
3983        route_take(&TABLE, 3, 9_100_000);
3984        route_reset(&TABLE);
3985        assert_eq!(CELL.route.load(Ordering::Relaxed), 4);
3986        route_unbind(&TABLE, 3);
3987        assert_eq!(CELL.route.load(Ordering::Relaxed), 0);
3988        // Wrapping: the stamp is the clock's low 32 bits.
3989        route_bind(&TABLE, "/in", 3);
3990        route_take(&TABLE, 3, u32::MAX as u64 - 9_999);
3991        let b = snapshot();
3992        CELL.count.fetch_add(1, Ordering::Relaxed);
3993        assert!(
3994            judge(u32::MAX as u64 + 90_001, &b).is_empty(),
3995            "100 ms across the wrap"
3996        );
3997    }
3998
3999    #[test]
4000    fn route_rule_code_is_appended() {
4001        assert_eq!(rule_code("route-latency-runtime"), 10);
4002        assert_eq!(rule_code("release-jitter-runtime"), 9);
4003        assert_eq!(rule_code("route-deadline-runtime"), 11);
4004    }
4005
4006    static DL_CELL: PubMonitorCell = PubMonitorCell::new();
4007    static DL_PLAIN: PubMonitorCell = PubMonitorCell::new();
4008    static DL_TABLE: [MonitorSpec; 2] = [
4009        // A route with the budget only: judged on completion, never overdue.
4010        MonitorSpec {
4011            topic: "/plain",
4012            fqn: "/n/plain",
4013            min_rate_hz_milli: 0,
4014            max_latency_ms: 0,
4015            route_trigger: "/in",
4016            route_max_latency_ms: 149,
4017            route_deadline_ms: 0,
4018            reaction_id: 0,
4019            mechanism: 0,
4020            cell: &DL_PLAIN,
4021        },
4022        // phase-503 -- the same route, supervised.
4023        MonitorSpec {
4024            topic: "/out",
4025            fqn: "/n/out",
4026            min_rate_hz_milli: 0,
4027            max_latency_ms: 0,
4028            route_trigger: "/in",
4029            route_max_latency_ms: 149,
4030            route_deadline_ms: 149,
4031            reaction_id: 3,
4032            mechanism: MECHANISM_TIMER,
4033            cell: &DL_CELL,
4034        },
4035    ];
4036
4037    fn overdue(now_us: u64) -> Vec<(usize, u32, u32, bool)> {
4038        let mut out = Vec::new();
4039        route_overdue(&DL_TABLE, now_us, |row, v, first| {
4040            assert_eq!(v.rule, "route-deadline-runtime");
4041            assert_eq!(v.fqn, "/n/out");
4042            out.push((row, v.measured, v.declared, first));
4043        });
4044        out
4045    }
4046
4047    /// phase-503 -- an output that never comes is judged at its deadline, not
4048    /// never: the wake source names the due time, the check at due reports it
4049    /// once with the row the hook is called for, and the next take starts a
4050    /// fresh route.
4051    #[test]
4052    fn route_deadline_rule() {
4053        assert!(route_deadline_active(&DL_TABLE));
4054        assert!(
4055            !route_deadline_active(&TABLE),
4056            "phase10-B3 rows stay unsupervised"
4057        );
4058        assert_eq!(route_next_due_us(&DL_TABLE, 0), None, "nothing pending");
4059        route_bind(&DL_TABLE, "/in", 2);
4060        route_take(&DL_TABLE, 2, 1_000_000);
4061        // The wake source: 149 ms after the take, then counting down.
4062        assert_eq!(route_next_due_us(&DL_TABLE, 1_000_000), Some(149_000));
4063        assert_eq!(route_next_due_us(&DL_TABLE, 1_100_000), Some(49_000));
4064        // One microsecond early: not overdue, still pending.
4065        assert!(overdue(1_148_999).is_empty());
4066        assert_ne!(DL_CELL.route.load(Ordering::Relaxed) & ROUTE_PENDING, 0);
4067        assert_eq!(route_next_due_us(&DL_TABLE, 1_148_999), Some(1));
4068        // At the deadline plus one 1 ms tick: reported for row 1, first.
4069        assert_eq!(overdue(1_150_000), [(1, 150, 149, true)]);
4070        let w = DL_CELL.route.load(Ordering::Relaxed);
4071        assert_eq!(w & ROUTE_PENDING, 0, "the take is dropped");
4072        assert_ne!(w & ROUTE_VIOLATED, 0);
4073        // The unsupervised row keeps its take for the completion verdict.
4074        assert_ne!(DL_PLAIN.route.load(Ordering::Relaxed) & ROUTE_PENDING, 0);
4075        assert_eq!(route_next_due_us(&DL_TABLE, 1_150_000), None);
4076        assert!(overdue(5_000_000).is_empty(), "judged once per take");
4077        // Still never answered: the next take misses too. The reaction is
4078        // requested again, the violation is not recorded twice.
4079        route_take(&DL_TABLE, 2, 6_000_000);
4080        assert_eq!(overdue(6_200_000), [(1, 200, 149, false)]);
4081        // A take answered in time clears the mark (route_judge's recovery).
4082        route_take(&DL_TABLE, 2, 7_000_000);
4083        let mut before = [0u32; MAX_MONITORS];
4084        for (k, s) in DL_TABLE.iter().enumerate() {
4085            before[k] = s.cell.count.load(Ordering::Relaxed);
4086        }
4087        DL_CELL.count.fetch_add(1, Ordering::Relaxed);
4088        let mut judged = Vec::new();
4089        route_judge(&DL_TABLE, &before, 7_010_000, |_, v, _, _| {
4090            judged.push(v.measured)
4091        });
4092        assert!(judged.is_empty(), "10 ms: within");
4093        assert_eq!(DL_CELL.route.load(Ordering::Relaxed) & ROUTE_VIOLATED, 0);
4094        // Over again after recovery: first again.
4095        route_take(&DL_TABLE, 2, 8_000_000);
4096        assert_eq!(overdue(8_149_000), [(1, 149, 149, true)]);
4097        // Across the clock's 32-bit wrap.
4098        route_take(&DL_TABLE, 2, u32::MAX as u64 - 9_999);
4099        assert_eq!(
4100            route_next_due_us(&DL_TABLE, u32::MAX as u64 + 1),
4101            Some(139_000)
4102        );
4103        assert!(overdue(u32::MAX as u64 + 139_000).is_empty());
4104        assert_eq!(overdue(u32::MAX as u64 + 139_001).len(), 1);
4105    }
4106
4107    static LC_CELL: PubMonitorCell = PubMonitorCell::new();
4108    static LC_TABLE: [MonitorSpec; 1] = [MonitorSpec {
4109        topic: "/out",
4110        fqn: "/n/late",
4111        min_rate_hz_milli: 0,
4112        max_latency_ms: 0,
4113        route_trigger: "/in",
4114        route_max_latency_ms: 149,
4115        route_deadline_ms: 149,
4116        reaction_id: 5,
4117        mechanism: MECHANISM_EXECUTOR,
4118        cell: &LC_CELL,
4119    }];
4120
4121    /// `(row, rule, measured, declared, record, react)` of one `route_judge`
4122    /// call after the output was published once.
4123    #[allow(clippy::type_complexity)]
4124    fn complete(end_us: u64) -> Vec<(usize, &'static str, u32, u32, bool, bool)> {
4125        let mut before = [0u32; MAX_MONITORS];
4126        before[0] = LC_CELL.count.load(Ordering::Relaxed);
4127        LC_CELL.count.fetch_add(1, Ordering::Relaxed);
4128        let mut out = Vec::new();
4129        route_judge(&LC_TABLE, &before, end_us, |row, v, record, react| {
4130            assert_eq!(v.fqn, "/n/late");
4131            out.push((row, v.rule, v.measured, v.declared, record, react));
4132        });
4133        out
4134    }
4135
4136    fn lc_overdue(now_us: u64) -> usize {
4137        let mut n = 0;
4138        route_overdue(&LC_TABLE, now_us, |_, _, _| n += 1);
4139        n
4140    }
4141
4142    /// phase-503 -- a supervised route whose one long callback publishes the
4143    /// output after the deadline, before the executor's overdue check could
4144    /// run: the completion reports the deadline miss and requests the
4145    /// reaction, once per take; a deadline already caught by the check is not
4146    /// reported again at completion; an on-time completion reports nothing.
4147    #[test]
4148    fn a_late_completion_is_a_deadline_miss_once() {
4149        route_bind(&LC_TABLE, "/in", 0);
4150        // Late completion, no overdue check in between (the callback held the
4151        // spin thread): the latency verdict, plus one deadline verdict that
4152        // asks for the reaction.
4153        route_take(&LC_TABLE, 0, 1_000_000);
4154        assert_eq!(
4155            complete(1_327_000),
4156            [
4157                (0, "route-latency-runtime", 327, 149, true, false),
4158                (0, "route-deadline-runtime", 327, 149, true, true),
4159            ]
4160        );
4161        assert_eq!(LC_CELL.route.load(Ordering::Relaxed) & ROUTE_PENDING, 0);
4162        assert_ne!(LC_CELL.route.load(Ordering::Relaxed) & ROUTE_VIOLATED, 0);
4163        // The take was answered: the overdue check has nothing left.
4164        assert_eq!(lc_overdue(1_400_000), 0, "no second verdict");
4165        // Late again while still marked: the reaction again, no record.
4166        route_take(&LC_TABLE, 0, 2_000_000);
4167        assert_eq!(
4168            complete(2_200_000),
4169            [(0, "route-deadline-runtime", 200, 149, false, true)]
4170        );
4171        // On time: nothing, and the mark clears.
4172        route_take(&LC_TABLE, 0, 3_000_000);
4173        assert!(complete(3_100_000).is_empty());
4174        assert_eq!(LC_CELL.route.load(Ordering::Relaxed) & ROUTE_VIOLATED, 0);
4175        // The deadline already fired (the check ran in time, or the timer
4176        // mechanism's): the late output that follows judges nothing.
4177        route_take(&LC_TABLE, 0, 4_000_000);
4178        assert_eq!(lc_overdue(4_150_000), 1);
4179        assert!(
4180            complete(4_327_000).is_empty(),
4181            "no second call at completion"
4182        );
4183        // Exactly at the deadline counts as overdue, as in route_overdue;
4184        // within the budget and before it does not.
4185        route_take(&LC_TABLE, 0, 5_000_000);
4186        let at = complete(5_149_000);
4187        assert_eq!(at.len(), 1);
4188        assert_eq!(at[0].1, "route-deadline-runtime");
4189        route_take(&LC_TABLE, 0, 6_000_000);
4190        assert!(complete(6_148_999).is_empty());
4191    }
4192}
4193
4194/// phase-462 W5b -- the three admission outcomes, decided from the backend's
4195/// own mask, and the arrival evidence an armed row reads.
4196#[cfg(test)]
4197mod unserved_deadline_tests {
4198    use super::*;
4199    use nros_rmw::{QoSPolicyMask, QoSProfile};
4200
4201    static CELL: SubMonitorCell = SubMonitorCell::new();
4202    static ROWS: [AgeMonitorSpec; 1] = [AgeMonitorSpec {
4203        topic: "/scan",
4204        fqn: "/h/scan",
4205        max_age_ms: 0,
4206        deadline_ms: 50,
4207        cell: &CELL,
4208    }];
4209    const ALL: QoSPolicyMask = QoSPolicyMask(u32::MAX);
4210    const NO_DEADLINE: QoSPolicyMask = QoSPolicyMask(!QoSPolicyMask::DEADLINE.0);
4211
4212    fn with_deadline(ms: u32) -> QoSProfile {
4213        QoSProfile {
4214            deadline_ms: ms,
4215            ..QoSProfile::default()
4216        }
4217    }
4218
4219    #[test]
4220    fn a_backend_that_honours_a_deadline_it_was_given_judges_it() {
4221        let (qos, row) = admit_contract_deadline(with_deadline(50), ALL, &ROWS, "/scan");
4222        assert_eq!((qos.deadline_ms, row.is_none()), (50, true));
4223    }
4224
4225    #[test]
4226    fn an_unserved_contract_deadline_is_stripped_and_armed() {
4227        let (qos, row) = admit_contract_deadline(with_deadline(50), NO_DEADLINE, &ROWS, "/scan");
4228        assert_eq!(qos.deadline_ms, 0, "stripped, so create does not refuse it");
4229        assert_eq!(row.map(|r| r.fqn), Some("/h/scan"));
4230    }
4231
4232    #[test]
4233    fn a_deadline_never_handed_to_the_backend_is_armed() {
4234        let (qos, row) = admit_contract_deadline(QoSProfile::default(), ALL, &ROWS, "/scan");
4235        assert_eq!((qos.deadline_ms, row.is_some()), (0, true));
4236    }
4237
4238    /// The negative controls: the caller's OWN different deadline is left for
4239    /// the backend to refuse by name, and a topic with no row is untouched.
4240    #[test]
4241    fn a_callers_own_deadline_and_an_uncontracted_topic_are_left_alone() {
4242        let (qos, _) = admit_contract_deadline(with_deadline(20), NO_DEADLINE, &ROWS, "/scan");
4243        assert_eq!(qos.deadline_ms, 20);
4244        let (qos, row) = admit_contract_deadline(with_deadline(50), NO_DEADLINE, &ROWS, "/other");
4245        assert_eq!((qos.deadline_ms, row.is_none()), (50, true));
4246    }
4247
4248    #[test]
4249    fn readiness_credits_only_the_armed_slot_and_unbind_disarms() {
4250        static C: SubMonitorCell = SubMonitorCell::new();
4251        static T: [AgeMonitorSpec; 1] = [AgeMonitorSpec {
4252            topic: "/t",
4253            fqn: "/n/t",
4254            max_age_ms: 0,
4255            deadline_ms: 10,
4256            cell: &C,
4257        }];
4258        credit_arrivals(&T, 1 << 3);
4259        assert_eq!(C.arrivals.load(Ordering::Relaxed), 0, "not armed");
4260        C.deadline_slot.store(4, Ordering::Relaxed);
4261        credit_arrivals(&T, 1 << 2);
4262        credit_arrivals(&T, 1 << 3);
4263        assert_eq!(C.arrivals.load(Ordering::Relaxed), 1);
4264        deadline_unbind(&T, 2);
4265        assert_eq!(C.deadline_slot.load(Ordering::Relaxed), 4, "another slot");
4266        deadline_unbind(&T, 3);
4267        assert_eq!(C.deadline_slot.load(Ordering::Relaxed), 0);
4268    }
4269
4270    /// An armed deadline-only row judges arrivals over the deadline; fed, it
4271    /// stays quiet. Unarmed, the same row judges nothing at all.
4272    #[test]
4273    fn an_armed_deadline_row_judges_arrivals_over_its_window() {
4274        static C: SubMonitorCell = SubMonitorCell::new();
4275        let spec = AgeMonitorSpec {
4276            topic: "/t",
4277            fqn: "/n/t",
4278            max_age_ms: 0,
4279            deadline_ms: 10,
4280            cell: &C,
4281        };
4282        let mut st = AgeState::default();
4283        assert!(check_age(&spec, &mut st, Some(1_000_000)).is_none());
4284        assert!(
4285            check_age(&spec, &mut st, Some(1_050_000)).is_none(),
4286            "unarmed"
4287        );
4288        C.deadline_slot.store(1, Ordering::Relaxed);
4289        let mut st = AgeState::default();
4290        assert!(
4291            check_age(&spec, &mut st, Some(2_000_000)).is_none(),
4292            "opens"
4293        );
4294        C.arrivals.store(1, Ordering::Relaxed);
4295        assert!(check_age(&spec, &mut st, Some(2_020_000)).is_none(), "fed");
4296        let v = check_age(&spec, &mut st, Some(2_040_000)).expect("silent past 10 ms");
4297        assert_eq!((v.rule, v.declared), ("silence-runtime", 10));
4298    }
4299}