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).
22//! - `deadline-miss-runtime` — a dispatched callback ran past its bound
23//!   SchedContext's `deadline_us`; what ELSE happens is the tier's
24//!   [`DeadlineAction`](super::sched_context::DeadlineAction).
25
26use core::sync::atomic::Ordering;
27// portable-atomic: RMW ops (fetch_add/fetch_max/swap) exist even on
28// riscv32imc / Cortex-M0+ that lack native CAS (same choice as
29// `SporadicState` / `AtomicSporadicState` in sched_context.rs).
30use portable_atomic::AtomicU32;
31
32/// One contracted publisher's counters. Baked as a `static` by codegen
33/// (or declared by the fixture); the publisher handle bumps `count` on
34/// every publish, the executor reads deltas on spin ticks.
35#[derive(Debug, Default)]
36pub struct PubMonitorCell {
37    pub count: AtomicU32,
38    /// W3b.5 — max observed take→publish latency (µs) in the current
39    /// check window. Written by the dispatch loop (fetch_max), drained
40    /// (swap 0) by the latency check.
41    pub max_latency_us: AtomicU32,
42}
43
44impl PubMonitorCell {
45    pub const fn new() -> Self {
46        Self {
47            count: AtomicU32::new(0),
48            max_latency_us: AtomicU32::new(0),
49        }
50    }
51}
52
53/// One contracted subscriber's take-age accumulator (W3b.5). The take
54/// path records `epoch_now - header.stamp` per message (fetch_max); the
55/// age check drains it (swap 0) per window.
56#[derive(Debug, Default)]
57pub struct SubMonitorCell {
58    /// Max observed take-age (ms) in the current check window.
59    pub max_age_ms: AtomicU32,
60}
61
62impl SubMonitorCell {
63    pub const fn new() -> Self {
64        Self {
65            max_age_ms: AtomicU32::new(0),
66        }
67    }
68
69    /// Take-path hook: record one message's age. `stamp_us` is the
70    /// peeked `header.stamp` as µs since the UNIX epoch, `epoch_now_us`
71    /// the receive-side wall clock. A stamp from the future clamps to 0.
72    pub fn observe(&self, stamp_us: u64, epoch_now_us: u64) {
73        let age_ms = (epoch_now_us.saturating_sub(stamp_us) / 1_000).min(u32::MAX as u64) as u32;
74        self.max_age_ms.fetch_max(age_ms, Ordering::Relaxed);
75    }
76}
77
78/// Peek `Time { i32 sec; u32 nanosec }` little-endian at `offset` in a
79/// raw CDR receive buffer (encapsulation header included) and return µs
80/// since the UNIX epoch. `None` when the buffer is too short or the
81/// stamp is pre-epoch/zero (unstamped messages never fire age monitors).
82pub fn peek_stamp_us(raw: &[u8], offset: usize) -> Option<u64> {
83    let sec_b = raw.get(offset..offset + 4)?;
84    let nsec_b = raw.get(offset + 4..offset + 8)?;
85    let sec = i32::from_le_bytes([sec_b[0], sec_b[1], sec_b[2], sec_b[3]]);
86    let nsec = u32::from_le_bytes([nsec_b[0], nsec_b[1], nsec_b[2], nsec_b[3]]);
87    if sec <= 0 {
88        return None;
89    }
90    Some(sec as u64 * 1_000_000 + nsec as u64 / 1_000)
91}
92
93/// One monitored publisher endpoint.
94#[derive(Debug, Clone, Copy)]
95pub struct MonitorSpec {
96    /// Topic name EXACTLY as the node passes it to `create_publisher`
97    /// (the SystemModel's wiring carries the same resolved name).
98    pub topic: &'static str,
99    /// Endpoint ref for violation reports (`<node FQN>/<endpoint>` — the
100    /// SystemModel contract key).
101    pub fqn: &'static str,
102    /// Declared publisher guarantee, in milli-Hz (fixed point: Hz × 1000).
103    /// 0 = no rate contract on this endpoint.
104    pub min_rate_hz_milli: u32,
105    /// W3b.5 — node-path budget (ms) for paths whose OUTPUT is this
106    /// endpoint (`contracts.node_paths[..].max_latency_ms`). 0 = no
107    /// latency contract.
108    pub max_latency_ms: u32,
109    /// The endpoint's counter cell.
110    pub cell: &'static PubMonitorCell,
111}
112
113/// One monitored subscriber endpoint (W3b.5 age contracts). Separate
114/// table from [`MonitorSpec`] — sub contracts key different endpoints
115/// and need no publish counter.
116#[derive(Debug, Clone, Copy)]
117pub struct AgeMonitorSpec {
118    /// Topic name EXACTLY as the node passes it to `create_subscription`.
119    pub topic: &'static str,
120    /// Endpoint ref for violation reports (the SystemModel contract key).
121    pub fqn: &'static str,
122    /// Declared max take-age (ms). 0 = no age contract.
123    pub max_age_ms: u32,
124    /// The endpoint's age accumulator.
125    pub cell: &'static SubMonitorCell,
126}
127
128/// Rate-check window (µs). Matches play_launch's ~5 s time-based trigger
129/// so both runtimes converge on comparable cadence.
130pub const RATE_CHECK_INTERVAL_US: u64 = 5_000_000;
131
132/// Max monitored endpoints per executor (const table, no_std).
133pub const MAX_MONITORS: usize = 8;
134/// Violation ring depth.
135pub const MAX_VIOLATIONS: usize = 8;
136
137/// A detected contract violation, in the play_launch rule-id vocabulary.
138#[derive(Debug, Clone)]
139pub struct Violation {
140    /// `"rate-hierarchy-runtime"` | `"max-age-runtime"` |
141    /// `"max-latency-runtime"` | `"deadline-miss-runtime"`.
142    pub rule: &'static str,
143    /// Violating endpoint ref (from the spec's `fqn`; the SC name for
144    /// deadline misses).
145    pub fqn: &'static str,
146    /// Measured value. Unit is per-rule: milli-Hz for the rate rule, ms
147    /// for age/latency, µs for deadline misses.
148    pub measured: u32,
149    /// Declared bound, same unit as `measured`.
150    pub declared: u32,
151}
152
153/// Per-spec accounting state (parallel to the spec table).
154#[derive(Debug, Clone, Copy, Default)]
155pub(crate) struct MonitorState {
156    /// Window opened (a plain bool, not a 0-sentinel on the timestamp —
157    /// `now_us == 0` is a legitimate first sample on freshly-started
158    /// monotonic clocks).
159    pub(crate) opened: bool,
160    pub(crate) window_start_us: u64,
161    pub(crate) count_at_window_start: u32,
162    /// Suppress duplicate reports: only re-report after a clean window.
163    pub(crate) violated_last_window: bool,
164    /// W3b.5 — separate dedup for the latency rule on the same spec row.
165    pub(crate) latency_violated_last_window: bool,
166}
167
168/// Pure rate check over one window boundary. Returns `Some(violation)`
169/// when the window elapsed AND the measured rate is below the declared
170/// minimum (and we didn't already report last window).
171///
172/// Extracted from the executor so the math is unit-testable without a
173/// session: publish counting is injected via the cell, time via `now_us`.
174pub(crate) fn check_rate(
175    spec: &MonitorSpec,
176    state: &mut MonitorState,
177    now_us: u64,
178) -> Option<Violation> {
179    if spec.min_rate_hz_milli == 0 {
180        return None;
181    }
182    let count = spec.cell.count.load(Ordering::Relaxed);
183    if !state.opened {
184        // First observation: open the window, no verdict yet.
185        state.opened = true;
186        state.window_start_us = now_us;
187        state.count_at_window_start = count;
188        return None;
189    }
190    let window_us = now_us.saturating_sub(state.window_start_us);
191    if window_us < RATE_CHECK_INTERVAL_US {
192        return None;
193    }
194    let published = count.wrapping_sub(state.count_at_window_start) as u64;
195    // milli-Hz = published * 1e3 / window_s = published * 1e9 / window_us
196    let measured_milli_hz =
197        (published.saturating_mul(1_000_000_000) / window_us.max(1)).min(u32::MAX as u64) as u32;
198
199    // Roll the window.
200    state.window_start_us = now_us;
201    state.count_at_window_start = count;
202
203    if measured_milli_hz < spec.min_rate_hz_milli {
204        if state.violated_last_window {
205            return None; // still violated — already reported
206        }
207        state.violated_last_window = true;
208        Some(Violation {
209            rule: "rate-hierarchy-runtime",
210            fqn: spec.fqn,
211            measured: measured_milli_hz,
212            declared: spec.min_rate_hz_milli,
213        })
214    } else {
215        state.violated_last_window = false;
216        None
217    }
218}
219
220/// Pure latency check: drains the spec cell's window-max take→publish
221/// latency and fires when it exceeds the declared node-path budget.
222/// Same report-once-until-recovery semantics as the rate rule; runs on
223/// every monitor tick (the cell accumulates between ticks, so no window
224/// bookkeeping is needed — draining IS the window roll).
225pub(crate) fn check_latency(spec: &MonitorSpec, state: &mut MonitorState) -> Option<Violation> {
226    if spec.max_latency_ms == 0 {
227        return None;
228    }
229    let max_us = spec.cell.max_latency_us.swap(0, Ordering::Relaxed);
230    let max_ms = max_us / 1_000;
231    if max_ms > spec.max_latency_ms {
232        if state.latency_violated_last_window {
233            return None;
234        }
235        state.latency_violated_last_window = true;
236        Some(Violation {
237            rule: "max-latency-runtime",
238            fqn: spec.fqn,
239            measured: max_ms,
240            declared: spec.max_latency_ms,
241        })
242    } else {
243        // A quiet window (no dispatch attributed) also counts as clean —
244        // recovery resets the dedup like the rate rule's clean window.
245        state.latency_violated_last_window = false;
246        None
247    }
248}
249
250/// Per-age-spec dedup state.
251#[derive(Debug, Clone, Copy, Default)]
252pub(crate) struct AgeState {
253    pub(crate) violated_last_window: bool,
254}
255
256/// Pure age check: drains the sub cell's window-max take-age and fires
257/// when it exceeds the declared bound. Report-once-until-recovery.
258pub(crate) fn check_age(spec: &AgeMonitorSpec, state: &mut AgeState) -> Option<Violation> {
259    if spec.max_age_ms == 0 {
260        return None;
261    }
262    let max_ms = spec.cell.max_age_ms.swap(0, Ordering::Relaxed);
263    if max_ms > spec.max_age_ms {
264        if state.violated_last_window {
265            return None;
266        }
267        state.violated_last_window = true;
268        Some(Violation {
269            rule: "max-age-runtime",
270            fqn: spec.fqn,
271            measured: max_ms,
272            declared: spec.max_age_ms,
273        })
274    } else {
275        state.violated_last_window = false;
276        None
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283
284    static CELL: PubMonitorCell = PubMonitorCell::new();
285
286    fn spec(min_milli: u32) -> MonitorSpec {
287        MonitorSpec {
288            topic: "/chatter",
289            fqn: "/demo/talker/chatter",
290            min_rate_hz_milli: min_milli,
291            max_latency_ms: 0,
292            cell: &CELL,
293        }
294    }
295
296    #[test]
297    fn slow_publisher_fires_once_until_recovery() {
298        CELL.count.store(0, Ordering::Relaxed);
299        let s = spec(100_000); // 100 Hz declared
300        let mut st = MonitorState::default();
301
302        // t=0: opens window.
303        assert!(check_rate(&s, &mut st, 0).is_none());
304        // 5 publishes in 5 s = 1 Hz — violation.
305        CELL.count.store(5, Ordering::Relaxed);
306        let v = check_rate(&s, &mut st, RATE_CHECK_INTERVAL_US).expect("fires");
307        assert_eq!(v.rule, "rate-hierarchy-runtime");
308        assert_eq!(v.fqn, "/demo/talker/chatter");
309        assert_eq!(v.measured, 1_000);
310        assert_eq!(v.declared, 100_000);
311        // Still slow next window — suppressed (no re-report spam).
312        CELL.count.store(10, Ordering::Relaxed);
313        assert!(check_rate(&s, &mut st, 2 * RATE_CHECK_INTERVAL_US).is_none());
314        // Recovers (500 publishes in 5 s = 100 Hz) — clean window resets.
315        CELL.count.store(510, Ordering::Relaxed);
316        assert!(check_rate(&s, &mut st, 3 * RATE_CHECK_INTERVAL_US).is_none());
317        // Degrades again — fires again.
318        CELL.count.store(511, Ordering::Relaxed);
319        assert!(check_rate(&s, &mut st, 4 * RATE_CHECK_INTERVAL_US).is_some());
320    }
321
322    #[test]
323    fn compliant_and_uncontracted_stay_silent() {
324        static C2: PubMonitorCell = PubMonitorCell::new();
325        let s = MonitorSpec {
326            topic: "/t",
327            fqn: "/n/t",
328            min_rate_hz_milli: 500, // 0.5 Hz
329            max_latency_ms: 0,
330            cell: &C2,
331        };
332        let mut st = MonitorState::default();
333        assert!(check_rate(&s, &mut st, 0).is_none());
334        C2.count.store(5, Ordering::Relaxed); // 1 Hz measured ≥ 0.5 Hz declared
335        assert!(check_rate(&s, &mut st, RATE_CHECK_INTERVAL_US).is_none());
336
337        // min_rate 0 = uncontracted: never fires, no state.
338        let s0 = MonitorSpec {
339            topic: "/t",
340            fqn: "/n/t",
341            min_rate_hz_milli: 0,
342            max_latency_ms: 0,
343            cell: &C2,
344        };
345        let mut st0 = MonitorState::default();
346        assert!(check_rate(&s0, &mut st0, 10 * RATE_CHECK_INTERVAL_US).is_none());
347    }
348
349    #[test]
350    fn stale_take_fires_age_once_until_recovery() {
351        static SC: SubMonitorCell = SubMonitorCell::new();
352        let s = AgeMonitorSpec {
353            topic: "/scan",
354            fqn: "/perc/detector/scan",
355            max_age_ms: 100,
356            cell: &SC,
357        };
358        let mut st = AgeState::default();
359
360        // Fresh message: stamped 5 ms ago — silent.
361        SC.observe(1_000_000_000, 1_000_005_000);
362        assert!(check_age(&s, &mut st).is_none());
363        // Stale: 250 ms old — fires with the measured age.
364        SC.observe(1_000_000_000, 1_000_250_000);
365        let v = check_age(&s, &mut st).expect("fires");
366        assert_eq!(v.rule, "max-age-runtime");
367        assert_eq!(v.fqn, "/perc/detector/scan");
368        assert_eq!(v.measured, 250);
369        assert_eq!(v.declared, 100);
370        // Still stale next window — suppressed.
371        SC.observe(1_000_000_000, 1_000_300_000);
372        assert!(check_age(&s, &mut st).is_none());
373        // Recovers — clean window resets; stale again refires.
374        SC.observe(1_000_000_000, 1_000_010_000);
375        assert!(check_age(&s, &mut st).is_none());
376        SC.observe(1_000_000_000, 1_000_999_000);
377        assert!(check_age(&s, &mut st).is_some());
378    }
379
380    #[test]
381    fn peek_stamp_reads_le_time_and_rejects_unstamped() {
382        // Encapsulation header (4B) + sec=100 nsec=5000 at offset 4.
383        let mut raw = [0u8; 12];
384        raw[4..8].copy_from_slice(&100i32.to_le_bytes());
385        raw[8..12].copy_from_slice(&5_000u32.to_le_bytes());
386        assert_eq!(peek_stamp_us(&raw, 4), Some(100_000_005));
387        // Zero / negative sec = unstamped: no age sample.
388        assert_eq!(peek_stamp_us(&[0u8; 12], 4), None);
389        // Short buffer: no panic, no sample.
390        assert_eq!(peek_stamp_us(&raw[..8], 4), None);
391    }
392
393    #[test]
394    fn slow_path_fires_latency_once_until_recovery() {
395        static C3: PubMonitorCell = PubMonitorCell::new();
396        let s = MonitorSpec {
397            topic: "/cmd",
398            fqn: "/ctrl/control/cmd",
399            min_rate_hz_milli: 0,
400            max_latency_ms: 10,
401            cell: &C3,
402        };
403        let mut st = MonitorState::default();
404        // 4 ms dispatch — within budget.
405        C3.max_latency_us.store(4_000, Ordering::Relaxed);
406        assert!(check_latency(&s, &mut st).is_none());
407        assert_eq!(C3.max_latency_us.load(Ordering::Relaxed), 0, "drained");
408        // 25 ms dispatch — fires.
409        C3.max_latency_us.store(25_000, Ordering::Relaxed);
410        let v = check_latency(&s, &mut st).expect("fires");
411        assert_eq!(v.rule, "max-latency-runtime");
412        assert_eq!(v.measured, 25);
413        assert_eq!(v.declared, 10);
414        // Still slow — suppressed; recovery resets.
415        C3.max_latency_us.store(30_000, Ordering::Relaxed);
416        assert!(check_latency(&s, &mut st).is_none());
417        C3.max_latency_us.store(1_000, Ordering::Relaxed);
418        assert!(check_latency(&s, &mut st).is_none());
419        C3.max_latency_us.store(30_000, Ordering::Relaxed);
420        assert!(check_latency(&s, &mut st).is_some());
421    }
422}
nros_node/executor/monitor.rs

nros_node/executor/
monitor.rs
