1use core::time::Duration;
31
32use crate::model::{
33 MetricState, PressureId, PressureSignal, PressureSnapshot, PressureState, SystemSnapshot,
34 UnavailableReason,
35};
36
37use super::{Hysteresis, SignalReading, Thresholds, signals};
38
39const DISABLED_RULE: &str = "diagnostics are disabled in configuration";
41
42const RECENT_INTERVALS: usize = 9;
48
49#[derive(Clone, Debug)]
55pub struct PressureEngine {
56 thresholds: Thresholds,
57 trackers: Vec<Hysteresis>,
59 recent_intervals: [Duration; RECENT_INTERVALS],
74 recent_len: usize,
76 recent_next: usize,
78 observations: u64,
79 discontinuities: u64,
80}
81
82impl PressureEngine {
83 #[must_use]
85 pub fn new(thresholds: Thresholds) -> Self {
86 let thresholds = thresholds.sanitized();
87 Self {
88 trackers: PressureId::DISPLAY_ORDER
89 .iter()
90 .map(|_| Hysteresis::new(&thresholds))
91 .collect(),
92 thresholds,
93 recent_intervals: [Duration::ZERO; RECENT_INTERVALS],
94 recent_len: 0,
95 recent_next: 0,
96 observations: 0,
97 discontinuities: 0,
98 }
99 }
100
101 #[must_use]
103 pub const fn thresholds(&self) -> &Thresholds {
104 &self.thresholds
105 }
106
107 pub fn set_thresholds(&mut self, thresholds: Thresholds) {
114 self.thresholds = thresholds.sanitized();
115 self.trackers = PressureId::DISPLAY_ORDER
116 .iter()
117 .map(|_| Hysteresis::new(&self.thresholds))
118 .collect();
119 self.observations = 0;
120 }
121
122 pub fn reset(&mut self) {
129 for tracker in &mut self.trackers {
130 tracker.reset();
131 }
132 self.observations = 0;
133 }
134
135 #[must_use]
137 pub const fn observations(&self) -> u64 {
138 self.observations
139 }
140
141 #[must_use]
146 pub const fn discontinuities(&self) -> u64 {
147 self.discontinuities
148 }
149
150 #[must_use]
155 pub fn observe(&mut self, snapshot: &SystemSnapshot) -> PressureSnapshot {
156 if !self.thresholds.enabled {
157 return PressureSnapshot {
158 signals: PressureId::DISPLAY_ORDER
159 .iter()
160 .map(|&id| PressureSignal::unsupported(id, DISABLED_RULE))
161 .collect(),
162 psi: snapshot.pressure.psi,
163 };
164 }
165
166 if !snapshot.has_valid_interval() {
170 return self.warming_up_snapshot(snapshot);
171 }
172
173 if self.is_discontinuity(snapshot.elapsed) {
174 self.discontinuities = self.discontinuities.saturating_add(1);
175 self.reset();
176 }
177 self.record_interval(snapshot.elapsed);
178 self.observations = self.observations.saturating_add(1);
179
180 let signals = PressureId::DISPLAY_ORDER
181 .iter()
182 .map(|&id| self.signal(id, snapshot))
183 .collect();
184 PressureSnapshot {
185 signals,
186 psi: snapshot.pressure.psi,
187 }
188 }
189
190 fn is_discontinuity(&self, elapsed: Duration) -> bool {
193 let Some(reference) = self.reference_interval() else {
194 return false;
195 };
196 let limit =
197 Thresholds::intervals_as_seconds(reference, self.thresholds.discontinuity_intervals);
198 elapsed.as_secs_f64() > limit
199 }
200
201 fn reference_interval(&self) -> Option<Duration> {
207 if self.recent_len < 2 {
208 return None;
209 }
210 let mut window = self.recent_intervals;
211 let filled = window.get_mut(..self.recent_len)?;
212 filled.sort_unstable();
213 filled.get(self.recent_len / 2).copied()
214 }
215
216 fn record_interval(&mut self, elapsed: Duration) {
218 if let Some(slot) = self.recent_intervals.get_mut(self.recent_next) {
219 *slot = elapsed;
220 }
221 self.recent_next = self.recent_next.saturating_add(1) % RECENT_INTERVALS;
222 self.recent_len = self.recent_len.saturating_add(1).min(RECENT_INTERVALS);
223 }
224
225 fn signal(&mut self, id: PressureId, snapshot: &SystemSnapshot) -> PressureSignal {
227 let reading = signals::read(id, snapshot, &self.thresholds);
228 let slot = Self::slot(id);
229 let Some(tracker) = self.trackers.get_mut(slot) else {
230 return Self::signal_from(id, &reading, MetricState::WarmingUp, None);
234 };
235
236 let Some(&candidate) = reading.state.fresh() else {
237 tracker.reset();
240 return Self::signal_from(id, &reading, reading.state, None);
241 };
242
243 let state = tracker.observe(candidate, snapshot.elapsed);
244 let held_for = tracker.held_for();
245 Self::signal_from(id, &reading, state, held_for)
246 }
247
248 fn signal_from(
255 id: PressureId,
256 reading: &SignalReading,
257 state: MetricState<PressureState>,
258 held_for: Option<Duration>,
259 ) -> PressureSignal {
260 let severity = match state {
261 MetricState::Available(_) => reading.severity,
262 MetricState::WarmingUp => MetricState::WarmingUp,
263 MetricState::PermissionDenied => MetricState::PermissionDenied,
264 MetricState::Unsupported => MetricState::Unsupported,
265 MetricState::TemporarilyUnavailable(reason) => {
266 MetricState::TemporarilyUnavailable(reason)
267 }
268 MetricState::Stale { .. } => {
271 MetricState::TemporarilyUnavailable(UnavailableReason::NeedsSecondSample)
272 }
273 };
274 PressureSignal {
275 id,
276 state,
277 severity,
278 raw: reading.raw,
279 rule: reading.rule,
280 held_for,
281 }
282 }
283
284 fn warming_up_snapshot(&self, snapshot: &SystemSnapshot) -> PressureSnapshot {
286 PressureSnapshot {
287 signals: PressureId::DISPLAY_ORDER
288 .iter()
289 .map(|&id| {
290 let reading = signals::read(id, snapshot, &self.thresholds);
291 let state = match reading.state {
294 MetricState::Available(_) | MetricState::Stale { .. } => {
295 MetricState::WarmingUp
296 }
297 other => other,
298 };
299 Self::signal_from(id, &reading, state, None)
300 })
301 .collect(),
302 psi: snapshot.pressure.psi,
303 }
304 }
305
306 fn slot(id: PressureId) -> usize {
308 PressureId::DISPLAY_ORDER
309 .iter()
310 .position(|candidate| *candidate == id)
311 .unwrap_or(0)
312 }
313}
314
315impl Default for PressureEngine {
316 fn default() -> Self {
318 Self::new(Thresholds::default())
319 }
320}
321
322#[cfg(test)]
323mod tests {
324 use super::*;
325 use crate::diagnostics::fixtures::{
326 Timeline, set_cpu, set_disk_busy, set_memory, set_psi, snapshot,
327 };
328 use crate::model::UnavailableReason;
329
330 const TOTAL: u64 = 32 * 1024 * 1024 * 1024;
331
332 fn engine() -> PressureEngine {
333 PressureEngine::default()
334 }
335
336 fn cpu_state(radar: &PressureSnapshot) -> MetricState<PressureState> {
337 radar
338 .signal(PressureId::Cpu)
339 .map_or(MetricState::Unsupported, |signal| signal.state)
340 }
341
342 fn feed_cpu(engine: &mut PressureEngine, busy: f32, count: usize) -> PressureSnapshot {
344 let mut timeline = Timeline::new(Duration::from_secs(1));
345 let mut radar = PressureSnapshot::warming_up();
346 for _ in 0..=count {
347 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, busy));
348 radar = engine.observe(&snapshot);
349 }
350 radar
351 }
352
353 #[test]
363 fn a_forced_refresh_does_not_make_every_later_sample_a_discontinuity() {
364 let mut engine = engine();
365 let mut timeline = Timeline::new(Duration::from_secs(1));
366
367 for _ in 0..4 {
369 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
370 let _ = engine.observe(&snapshot);
371 }
372 let mut forced = timeline.build(|snapshot| set_cpu(snapshot, 99.0));
373 forced.elapsed = Duration::from_millis(3);
374 assert!(timeline.record(&forced));
375 let _ = engine.observe(&forced);
376 let after_forced = engine.discontinuities();
377
378 for _ in 0..20 {
380 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
381 let _ = engine.observe(&snapshot);
382 }
383
384 assert_eq!(
385 engine.discontinuities(),
386 after_forced,
387 "a 1s sample after a 3ms one is not a sleep/wake gap"
388 );
389 let radar = engine.observe(&timeline.push(|snapshot| set_cpu(snapshot, 99.0)));
390 let signal = radar.signal(PressureId::Cpu).expect("cpu signal exists");
391 assert_eq!(
392 signal.state,
393 MetricState::Available(PressureState::Critical),
394 "the radar must still be able to commit a state, got {:?}",
395 signal.state
396 );
397 }
398
399 #[test]
401 fn a_gap_far_larger_than_the_recent_cadence_is_still_a_discontinuity() {
402 let mut engine = engine();
403 let mut timeline = Timeline::new(Duration::from_secs(1));
404 for _ in 0..12 {
405 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
406 let _ = engine.observe(&snapshot);
407 }
408 let before = engine.discontinuities();
409
410 let mut resumed = timeline.build(|snapshot| set_cpu(snapshot, 99.0));
412 resumed.elapsed = Duration::from_secs(60);
413 assert!(timeline.record(&resumed));
414 let radar = engine.observe(&resumed);
415
416 assert_eq!(engine.discontinuities(), before + 1);
417 let signal = radar.signal(PressureId::Cpu).expect("cpu signal exists");
418 assert!(
419 signal.state.is_warming_up(),
420 "§11.3: the window either side of a gap must not be counted together, \
421 got {:?}",
422 signal.state
423 );
424 }
425
426 #[test]
427 fn the_radar_always_contains_every_signal_in_display_order() {
428 let radar = engine().observe(&snapshot());
429 assert_eq!(radar.signals.len(), PressureId::DISPLAY_ORDER.len());
430 for (signal, expected) in radar.signals.iter().zip(PressureId::DISPLAY_ORDER) {
431 assert_eq!(signal.id, expected);
432 assert!(!signal.rule.is_empty(), "§2.3 requires the rule text");
433 }
434 }
435
436 #[test]
437 fn the_first_snapshot_produces_no_state_at_all() {
438 let mut engine = engine();
439 let radar = engine.observe(&snapshot());
440 assert!(
441 radar.worst_state().is_warming_up(),
442 "an unmeasured system must not read as healthy"
443 );
444 assert_eq!(engine.observations(), 0, "a zero interval is not a sample");
445 }
446
447 #[test]
448 fn a_signal_warms_up_until_the_minimum_sample_count_is_reached() {
449 let mut engine = engine();
450 let mut timeline = Timeline::new(Duration::from_secs(1));
451 for index in 0..10 {
452 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
453 let radar = engine.observe(&snapshot);
454 let state = cpu_state(&radar);
455 if index == 0 {
456 assert!(state.is_warming_up(), "the first sample has no interval");
457 continue;
458 }
459 assert!(
460 state.is_warming_up(),
461 "sample {index} must not support a sustained claim"
462 );
463 }
464 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
465 assert_eq!(
466 cpu_state(&engine.observe(&snapshot)),
467 MetricState::Available(PressureState::Critical)
468 );
469 }
470
471 #[test]
472 fn a_sustained_condition_escalates_and_reports_how_long_it_has_held() {
473 let mut engine = engine();
474 let radar = feed_cpu(&mut engine, 99.0, 20);
475 let signal = radar.signal(PressureId::Cpu).expect("cpu signal exists");
476
477 assert_eq!(
478 signal.state,
479 MetricState::Available(PressureState::Critical)
480 );
481 assert_eq!(signal.symbol(), 'X', "§2.3's redundant cue");
482 assert!(signal.severity.fresh().is_some());
483 assert!(signal.raw.is_some(), "§2.3 requires the raw metric");
484 assert!(
485 signal
486 .held_for
487 .is_some_and(|held| held >= Duration::from_secs(10)),
488 "held_for {:?}",
489 signal.held_for
490 );
491 }
492
493 #[test]
494 fn an_alternating_metric_does_not_flap_the_radar() {
495 let mut engine = engine();
498 let mut timeline = Timeline::new(Duration::from_secs(1));
499 let mut states = Vec::new();
500 for index in 0..60 {
501 let busy = if index % 2 == 0 { 99.0 } else { 1.0 };
502 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, busy));
503 states.push(cpu_state(&engine.observe(&snapshot)));
504 }
505
506 let derived: Vec<PressureState> = states
507 .iter()
508 .filter_map(|state| state.fresh().copied())
509 .collect();
510 assert!(!derived.is_empty(), "the signal must settle eventually");
511 assert!(
512 derived.iter().all(|state| *state == PressureState::Normal),
513 "the radar flapped: {derived:?}"
514 );
515 }
516
517 #[test]
518 fn an_unavailable_input_leaves_the_signal_unavailable_rather_than_normal() {
519 let mut engine = engine();
520 feed_cpu(&mut engine, 99.0, 20);
521
522 let mut timeline = Timeline::new(Duration::from_secs(1));
523 let snapshot = timeline.push(|snapshot| {
524 snapshot.cpu.total = MetricState::PermissionDenied;
525 });
526 let radar = engine.observe(&snapshot);
527 assert_eq!(cpu_state(&radar), MetricState::PermissionDenied);
528 let signal = radar.signal(PressureId::Cpu).expect("cpu signal exists");
529 assert_eq!(signal.symbol(), '!');
530 assert!(signal.held_for.is_none());
531 }
532
533 #[test]
534 fn a_counter_reset_clears_the_window_instead_of_counting_as_an_event() {
535 let mut engine = engine();
536 let mut timeline = Timeline::new(Duration::from_secs(1));
537
538 for _ in 0..10 {
540 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
541 let _ = engine.observe(&snapshot);
542 }
543 let reset = timeline.push(|snapshot| {
544 snapshot.cpu.total =
545 MetricState::TemporarilyUnavailable(UnavailableReason::CounterReset);
546 });
547 assert_eq!(
548 cpu_state(&engine.observe(&reset)),
549 MetricState::TemporarilyUnavailable(UnavailableReason::CounterReset)
550 );
551
552 let after = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
553 assert!(
554 cpu_state(&engine.observe(&after)).is_warming_up(),
555 "§11.3: a reset must not be readable as an event"
556 );
557 }
558
559 #[test]
560 fn a_sleep_wake_gap_resets_every_signal() {
561 let mut engine = engine();
562 let mut timeline = Timeline::new(Duration::from_secs(1));
563 for _ in 0..15 {
564 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
565 let _ = engine.observe(&snapshot);
566 }
567 assert_eq!(
568 cpu_state(&engine.observe(&timeline.push(|s| set_cpu(s, 99.0)))),
569 MetricState::Available(PressureState::Critical)
570 );
571
572 let mut woken = timeline.build(|snapshot| set_cpu(snapshot, 99.0));
574 woken.elapsed = Duration::from_secs(7_200);
575 let radar = engine.observe(&woken);
576
577 assert!(
578 cpu_state(&radar).is_warming_up(),
579 "the gap must not be read as fifteen saturated samples"
580 );
581 assert_eq!(engine.discontinuities(), 1);
582 }
583
584 #[test]
585 fn a_shorter_than_usual_interval_is_not_a_discontinuity() {
586 let mut engine = engine();
587 let mut timeline = Timeline::new(Duration::from_secs(1));
588 for _ in 0..12 {
589 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
590 let _ = engine.observe(&snapshot);
591 }
592 let mut jittered = timeline.build(|snapshot| set_cpu(snapshot, 99.0));
593 jittered.elapsed = Duration::from_millis(600);
594 let _ = engine.observe(&jittered);
595 assert_eq!(engine.discontinuities(), 0);
596 }
597
598 #[test]
599 fn the_reference_interval_cannot_be_inflated_by_a_stall() {
600 let mut engine = engine();
601 let mut timeline = Timeline::new(Duration::from_secs(1));
602 for _ in 0..3 {
603 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 10.0));
604 let _ = engine.observe(&snapshot);
605 }
606 let mut stalled = timeline.build(|snapshot| set_cpu(snapshot, 10.0));
608 stalled.elapsed = Duration::from_secs(5);
609 let _ = engine.observe(&stalled);
610 assert_eq!(engine.discontinuities(), 0);
611
612 let mut gap = timeline.build(|snapshot| set_cpu(snapshot, 10.0));
614 gap.elapsed = Duration::from_secs(30);
615 let _ = engine.observe(&gap);
616 assert_eq!(engine.discontinuities(), 1);
617 }
618
619 #[test]
620 fn raw_psi_is_carried_through_untouched() {
621 let mut engine = engine();
622 let mut snapshot = snapshot();
623 set_psi(&mut snapshot, 1.0, 2.0, 3.0);
624 let radar = engine.observe(&snapshot);
625 assert_eq!(
626 radar.psi, snapshot.pressure.psi,
627 "psi is the collector's measurement, not the engine's"
628 );
629 }
630
631 #[test]
632 fn signals_the_platform_cannot_measure_stay_unsupported_forever() {
633 let mut engine = engine();
636 let mut timeline = Timeline::new(Duration::from_secs(1));
637 for _ in 0..20 {
638 let snapshot = timeline.push(|snapshot| {
639 set_cpu(snapshot, 10.0);
640 snapshot.pressure.psi = MetricState::Unsupported;
641 });
642 let radar = engine.observe(&snapshot);
643 for id in [PressureId::PsiCpu, PressureId::PsiMemory, PressureId::PsiIo] {
644 let signal = radar.signal(id).expect("signal exists");
645 assert!(
646 signal.state.is_unsupported(),
647 "{id:?} became {:?} without PSI data",
648 signal.state
649 );
650 assert_eq!(signal.symbol(), '-');
651 }
652 }
653 }
654
655 #[test]
656 fn a_metric_the_collector_has_not_reported_yet_stays_warming_up() {
657 let mut engine = engine();
661 let mut timeline = Timeline::new(Duration::from_secs(1));
662 for _ in 0..20 {
663 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 10.0));
664 let radar = engine.observe(&snapshot);
665 let signal = radar.signal(PressureId::PsiMemory).expect("signal exists");
666 assert!(signal.state.is_warming_up(), "{:?}", signal.state);
667 assert!(signal.state.fresh().is_none());
668 }
669 }
670
671 #[test]
672 fn independent_signals_do_not_share_hysteresis_state() {
673 let mut engine = engine();
674 let mut timeline = Timeline::new(Duration::from_secs(1));
675 for _ in 0..12 {
676 let snapshot = timeline.push(|snapshot| {
677 set_cpu(snapshot, 99.0);
678 set_memory(snapshot, TOTAL, TOTAL / 2);
679 set_disk_busy(snapshot, "nvme0n1", 5.0);
680 });
681 let radar = engine.observe(&snapshot);
682 let _ = radar;
683 }
684 let snapshot = timeline.push(|snapshot| {
685 set_cpu(snapshot, 99.0);
686 set_memory(snapshot, TOTAL, TOTAL / 2);
687 set_disk_busy(snapshot, "nvme0n1", 5.0);
688 });
689 let radar = engine.observe(&snapshot);
690
691 assert_eq!(
692 cpu_state(&radar),
693 MetricState::Available(PressureState::Critical)
694 );
695 assert_eq!(
696 radar.signal(PressureId::Memory).map(|signal| signal.state),
697 Some(MetricState::Available(PressureState::Normal))
698 );
699 assert_eq!(
700 radar.signal(PressureId::Disk).map(|signal| signal.state),
701 Some(MetricState::Available(PressureState::Normal))
702 );
703 }
704
705 #[test]
706 fn disabling_diagnostics_reports_unsupported_rather_than_healthy() {
707 let mut engine = PressureEngine::new(Thresholds {
708 enabled: false,
709 ..Thresholds::default()
710 });
711 let mut timeline = Timeline::new(Duration::from_secs(1));
712 for _ in 0..20 {
713 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
714 let radar = engine.observe(&snapshot);
715 for signal in &radar.signals {
716 assert!(signal.state.is_unsupported());
717 assert_eq!(signal.rule, DISABLED_RULE);
718 }
719 assert!(radar.worst_state().is_warming_up());
720 }
721 }
722
723 #[test]
724 fn changing_thresholds_restarts_the_evidence_rather_than_reusing_it() {
725 let mut engine = engine();
726 feed_cpu(&mut engine, 99.0, 20);
727
728 engine.set_thresholds(Thresholds {
729 cpu_watch_percent: 20.0,
730 ..Thresholds::default()
731 });
732 assert_eq!(engine.observations(), 0);
733
734 let mut timeline = Timeline::new(Duration::from_secs(1));
735 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
736 assert!(
737 cpu_state(&engine.observe(&snapshot)).is_warming_up(),
738 "observations made under other thresholds must not be reused"
739 );
740 }
741
742 #[test]
743 fn an_explicit_reset_returns_every_signal_to_warming_up() {
744 let mut engine = engine();
745 feed_cpu(&mut engine, 99.0, 20);
746 engine.reset();
747
748 let mut timeline = Timeline::new(Duration::from_secs(1));
749 timeline.push(|snapshot| set_cpu(snapshot, 99.0));
752 let snapshot = timeline.push(|snapshot| set_cpu(snapshot, 99.0));
753 let radar = engine.observe(&snapshot);
754 assert!(cpu_state(&radar).is_warming_up());
755 assert_eq!(engine.observations(), 1);
756 }
757
758 #[test]
759 fn the_worst_state_across_the_radar_is_what_the_header_shows() {
760 let mut engine = engine();
761 let mut timeline = Timeline::new(Duration::from_secs(1));
762 for _ in 0..12 {
763 let snapshot = timeline.push(|snapshot| {
764 set_cpu(snapshot, 10.0);
765 set_memory(snapshot, TOTAL, TOTAL / 100);
766 });
767 let _ = engine.observe(&snapshot);
768 }
769 let snapshot = timeline.push(|snapshot| {
770 set_cpu(snapshot, 10.0);
771 set_memory(snapshot, TOTAL, TOTAL / 100);
772 });
773 let radar = engine.observe(&snapshot);
774 assert_eq!(
775 radar.worst_state(),
776 MetricState::Available(PressureState::Critical),
777 "one critical signal makes the system critical"
778 );
779 }
780}