chio_supervisor/
health.rs

1//! The health flag: one monotonic, cloneable state handle per supervised surface.
2//!
3//! The governing rule is honesty. A supervisor that restarted its worker but lost
4//! work in the gap must still report the gap, so a tripped flag never clears itself
5//! on a later lucky success. The only path back to [`HealthLevel::Healthy`] is an
6//! explicit, operator-visible [`HealthFlag::clear`].
7
8use crate::sync::{Arc, AtomicU32, AtomicU64, AtomicU8, Mutex, Ordering};
9
10/// Severity of a supervised surface. Ordered by increasing severity; the flag only
11/// ever raises the level and never lowers it except through [`HealthFlag::clear`].
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
13#[serde(rename_all = "snake_case")]
14pub enum HealthLevel {
15    /// Serving normally.
16    #[default]
17    Healthy = 0,
18    /// Tripped after `trip_after` consecutive restarts. The surface reports degraded
19    /// and, if TCB-critical, fails evaluations closed. Restarts may still be running.
20    Degraded = 1,
21    /// Terminal: the restart budget is exhausted and the supervisor stopped respawning.
22    Failed = 2,
23}
24
25impl HealthLevel {
26    fn from_u8(value: u8) -> Self {
27        match value {
28            0 => HealthLevel::Healthy,
29            1 => HealthLevel::Degraded,
30            _ => HealthLevel::Failed,
31        }
32    }
33}
34
35/// A cloneable handle to one surface's health. Level and counter reads are
36/// lock-free; the reason string lives behind a `Mutex` whose only invariant is the
37/// single `Option` it holds, so recovering a poisoned lock with `into_inner` is
38/// sound (there is no cross-field state to leave half-mutated).
39#[derive(Clone)]
40pub struct HealthFlag(Arc<HealthState>);
41
42struct HealthState {
43    level: AtomicU8,
44    tcb_critical: bool,
45    consecutive_failures: AtomicU32,
46    restart_total: AtomicU64,
47    last_ok_unix_ms: AtomicU64,
48    last_transition_unix_ms: AtomicU64,
49    reason: Mutex<Option<String>>,
50}
51
52impl HealthFlag {
53    /// Create a flag in the [`HealthLevel::Healthy`] state. `tcb_critical` marks a
54    /// surface whose degradation must fail evaluations closed.
55    #[must_use]
56    pub fn new(tcb_critical: bool) -> Self {
57        Self(Arc::new(HealthState {
58            level: AtomicU8::new(HealthLevel::Healthy as u8),
59            tcb_critical,
60            consecutive_failures: AtomicU32::new(0),
61            restart_total: AtomicU64::new(0),
62            last_ok_unix_ms: AtomicU64::new(0),
63            last_transition_unix_ms: AtomicU64::new(0),
64            reason: Mutex::new(None),
65        }))
66    }
67
68    /// Record a completed unit of work. Resets the consecutive-failure counter and
69    /// stamps liveness, but never lowers a tripped level.
70    pub fn record_ok(&self, now_ms: u64) {
71        self.0.consecutive_failures.store(0, Ordering::SeqCst);
72        self.0.last_ok_unix_ms.store(now_ms, Ordering::SeqCst);
73    }
74
75    /// Record a restart-worthy failure and return the new consecutive count. Trips to
76    /// [`HealthLevel::Degraded`] once the count reaches `trip_after`, and never
77    /// downgrades. `trip_after == 0` trips on the first failure.
78    pub fn record_failure(&self, reason: impl Into<String>, now_ms: u64, trip_after: u32) -> u32 {
79        let count = self.0.consecutive_failures.fetch_add(1, Ordering::SeqCst) + 1;
80        self.0.restart_total.fetch_add(1, Ordering::SeqCst);
81        self.set_reason(Some(reason.into()));
82        if count >= trip_after {
83            self.raise_to(HealthLevel::Degraded, now_ms);
84        }
85        count
86    }
87
88    /// Terminal escalation: the restart budget is exhausted and the supervisor has
89    /// stopped respawning. Raises the level to [`HealthLevel::Failed`].
90    pub fn escalate_failed(&self, now_ms: u64) {
91        self.raise_to(HealthLevel::Failed, now_ms);
92    }
93
94    /// The current severity level.
95    #[must_use]
96    pub fn level(&self) -> HealthLevel {
97        HealthLevel::from_u8(self.0.level.load(Ordering::SeqCst))
98    }
99
100    /// Whether this surface is TCB-critical.
101    #[must_use]
102    pub fn tcb_critical(&self) -> bool {
103        self.0.tcb_critical
104    }
105
106    /// True when the surface must fail closed: TCB-critical and not
107    /// [`HealthLevel::Healthy`]. Read on the pre-dispatch path.
108    #[must_use]
109    pub fn is_serving_closed(&self) -> bool {
110        self.0.tcb_critical && !matches!(self.level(), HealthLevel::Healthy)
111    }
112
113    /// The only path back to [`HealthLevel::Healthy`]: an explicit operator recovery.
114    /// Clears the reason and resets the consecutive-failure counter; the cumulative
115    /// `restart_total` is preserved so the history of the incident is not erased.
116    pub fn clear(&self, now_ms: u64) {
117        self.0.consecutive_failures.store(0, Ordering::SeqCst);
118        self.0
119            .level
120            .store(HealthLevel::Healthy as u8, Ordering::SeqCst);
121        self.0
122            .last_transition_unix_ms
123            .store(now_ms, Ordering::SeqCst);
124        self.set_reason(None);
125    }
126
127    /// Point-in-time view for operator surfaces and telemetry exporters.
128    #[must_use]
129    pub fn snapshot(&self) -> HealthSnapshot {
130        HealthSnapshot {
131            level: self.level(),
132            tcb_critical: self.0.tcb_critical,
133            restart_total: self.0.restart_total.load(Ordering::SeqCst),
134            consecutive_failures: self.0.consecutive_failures.load(Ordering::SeqCst),
135            last_ok_unix_ms: nonzero(self.0.last_ok_unix_ms.load(Ordering::SeqCst)),
136            last_transition_unix_ms: nonzero(self.0.last_transition_unix_ms.load(Ordering::SeqCst)),
137            reason: self.reason(),
138        }
139    }
140
141    /// Monotonically raise the level. A compare-and-swap loop guarantees a concurrent
142    /// escalation can never be lost behind a lower-severity raise.
143    fn raise_to(&self, level: HealthLevel, now_ms: u64) {
144        let target = level as u8;
145        let mut current = self.0.level.load(Ordering::SeqCst);
146        while target > current {
147            match self
148                .0
149                .level
150                .compare_exchange(current, target, Ordering::SeqCst, Ordering::SeqCst)
151            {
152                Ok(_) => {
153                    self.0
154                        .last_transition_unix_ms
155                        .store(now_ms, Ordering::SeqCst);
156                    return;
157                }
158                Err(observed) => current = observed,
159            }
160        }
161    }
162
163    fn set_reason(&self, reason: Option<String>) {
164        match self.0.reason.lock() {
165            Ok(mut guard) => *guard = reason,
166            Err(poisoned) => *poisoned.into_inner() = reason,
167        }
168    }
169
170    fn reason(&self) -> Option<String> {
171        match self.0.reason.lock() {
172            Ok(guard) => guard.clone(),
173            Err(poisoned) => poisoned.into_inner().clone(),
174        }
175    }
176}
177
178/// A serializable point-in-time view of a [`HealthFlag`], for operator surfaces and
179/// the telemetry exporter. All fields are additive; consumers deserialize with
180/// `#[serde(default)]`.
181#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
182#[serde(rename_all = "camelCase")]
183pub struct HealthSnapshot {
184    pub level: HealthLevel,
185    pub tcb_critical: bool,
186    pub restart_total: u64,
187    pub consecutive_failures: u32,
188    #[serde(default)]
189    pub last_ok_unix_ms: Option<u64>,
190    #[serde(default)]
191    pub last_transition_unix_ms: Option<u64>,
192    #[serde(default)]
193    pub reason: Option<String>,
194}
195
196fn nonzero(value: u64) -> Option<u64> {
197    (value != 0).then_some(value)
198}
199
200#[cfg(all(test, not(loom)))]
201mod tests {
202    use super::*;
203
204    const NOW: u64 = 1_700_000_000_000;
205
206    #[test]
207    fn healthy_by_default_and_never_serves_closed_when_healthy() {
208        let flag = HealthFlag::new(true);
209        assert_eq!(flag.level(), HealthLevel::Healthy);
210        assert!(!flag.is_serving_closed());
211    }
212
213    #[test]
214    fn record_ok_resets_consecutive_failures_without_lowering_level() {
215        let flag = HealthFlag::new(false);
216        flag.record_failure("boom", NOW, 2);
217        flag.record_failure("boom", NOW, 2);
218        assert_eq!(flag.level(), HealthLevel::Degraded);
219        flag.record_ok(NOW + 1);
220        // The consecutive counter reset, but a tripped level must not self-heal.
221        assert_eq!(flag.snapshot().consecutive_failures, 0);
222        assert_eq!(flag.level(), HealthLevel::Degraded);
223    }
224
225    #[test]
226    fn trips_to_degraded_exactly_at_trip_after() {
227        let flag = HealthFlag::new(false);
228        assert_eq!(flag.record_failure("one", NOW, 3), 1);
229        assert_eq!(flag.level(), HealthLevel::Healthy);
230        assert_eq!(flag.record_failure("two", NOW, 3), 2);
231        assert_eq!(flag.level(), HealthLevel::Healthy);
232        assert_eq!(flag.record_failure("three", NOW, 3), 3);
233        assert_eq!(flag.level(), HealthLevel::Degraded);
234    }
235
236    #[test]
237    fn trip_after_zero_trips_on_first_failure() {
238        let flag = HealthFlag::new(false);
239        flag.record_failure("immediate", NOW, 0);
240        assert_eq!(flag.level(), HealthLevel::Degraded);
241    }
242
243    #[test]
244    fn escalate_failed_is_terminal_and_reported() {
245        let flag = HealthFlag::new(true);
246        flag.record_failure("boom", NOW, 1);
247        flag.escalate_failed(NOW);
248        assert_eq!(flag.level(), HealthLevel::Failed);
249        assert!(flag.is_serving_closed());
250    }
251
252    #[test]
253    fn raise_is_monotonic_and_ignores_lower_severity() {
254        let flag = HealthFlag::new(false);
255        flag.escalate_failed(NOW);
256        assert_eq!(flag.level(), HealthLevel::Failed);
257        // A later degraded trip must not lower a Failed surface.
258        flag.record_failure("late", NOW, 0);
259        assert_eq!(flag.level(), HealthLevel::Failed);
260    }
261
262    #[test]
263    fn tcb_flag_serves_closed_when_not_healthy_only() {
264        let non_tcb = HealthFlag::new(false);
265        non_tcb.escalate_failed(NOW);
266        assert!(!non_tcb.is_serving_closed());
267
268        let tcb = HealthFlag::new(true);
269        assert!(!tcb.is_serving_closed());
270        tcb.record_failure("boom", NOW, 0);
271        assert!(tcb.is_serving_closed());
272    }
273
274    #[test]
275    fn clear_is_the_only_downgrade_and_preserves_restart_total() {
276        let flag = HealthFlag::new(true);
277        flag.record_failure("boom", NOW, 1);
278        flag.record_failure("boom", NOW, 1);
279        assert_eq!(flag.level(), HealthLevel::Degraded);
280        let restarts_before = flag.snapshot().restart_total;
281        flag.clear(NOW + 5);
282        assert_eq!(flag.level(), HealthLevel::Healthy);
283        assert!(!flag.is_serving_closed());
284        assert_eq!(flag.snapshot().reason, None);
285        // History is not erased: the cumulative restart count survives the recovery.
286        assert_eq!(flag.snapshot().restart_total, restarts_before);
287    }
288
289    #[test]
290    fn snapshot_reports_reason_and_liveness() {
291        let flag = HealthFlag::new(false);
292        assert_eq!(flag.snapshot().last_ok_unix_ms, None);
293        flag.record_ok(NOW);
294        flag.record_failure("disk full", NOW + 1, 0);
295        let snap = flag.snapshot();
296        assert_eq!(snap.level, HealthLevel::Degraded);
297        assert_eq!(snap.restart_total, 1);
298        assert_eq!(snap.last_ok_unix_ms, Some(NOW));
299        assert_eq!(snap.reason.as_deref(), Some("disk full"));
300    }
301
302    #[test]
303    fn snapshot_round_trips_through_json() {
304        let flag = HealthFlag::new(true);
305        flag.record_failure("boom", NOW, 0);
306        let snap = flag.snapshot();
307        let encoded = serde_json::to_string(&snap).unwrap_or_default();
308        assert!(encoded.contains("\"level\":\"degraded\""));
309        let decoded: HealthSnapshot =
310            serde_json::from_str(&encoded).unwrap_or_else(|_| snap.clone());
311        assert_eq!(decoded, snap);
312    }
313}
chio_supervisor/health.rs

chio_supervisor/
health.rs
