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}