planarian_schema/
state.rs

1//! Observed state, derived status, and the state digest (§8).
2
3use crate::ids::{GenerationId, Hash, ProbeId, Timestamp};
4use crate::unit::{Condition, Evidence, Override, OverrideDesired, ProbeResult, ProbeRole, UnitDefs, UnitId, UnitKind};
5use serde::{Deserialize, Serialize};
6use std::collections::{BTreeMap, BTreeSet};
7
8#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
9pub enum Observed {
10    /// Stale evidence, probe timeout, backend error.
11    Unknown {
12        since: Timestamp,
13        reason: String,
14    },
15    Transitioning {
16        from: Box<Observed>,
17        since: Timestamp,
18    },
19    Absent,
20    Detected,
21    Bound,
22    Active,
23    Failed(Reason),
24}
25
26#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
27pub enum Reason {
28    ExitCode(i32),
29    DriverError(String),
30    Backend(String),
31}
32
33/// Derived, never stored as truth on its own.
34///
35/// `NotExpected` is not in §8. §8 says a unit whose desired condition does not
36/// hold reports its observed status and opens no incident. Mapping such a unit
37/// onto `Running` would break I4, so it gets a variant of its own.
38#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
39pub enum UnitStatus {
40    Running,
41    Degraded(Vec<ProbeId>),
42    Stopped,
43    Overridden(OverrideDesired),
44    Unknown,
45    Failed(Reason),
46    Absent,
47    Detected,
48    Bound,
49    Transitioning,
50    NotExpected,
51}
52
53/// Host resources. Excluded from the digest.
54#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
55pub struct Resources {
56    pub mem_total_mb: u64,
57    pub mem_available_mb: u64,
58    /// Free bytes per storage unit (`device:fs-root`, ...).
59    pub fs_free_bytes: BTreeMap<UnitId, u64>,
60}
61
62#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
63pub struct SystemState {
64    pub generation: GenerationId,
65    /// Boot-confirmed.
66    pub known_good: GenerationId,
67    /// Passed probes live, not yet rebooted.
68    pub live_tested: Option<GenerationId>,
69    /// Set by `mark-good --accept-unconfirmed`; never replaces `known_good` in the boot menu (§16, G3).
70    pub accepted_baseline: Option<GenerationId>,
71    /// §12e: derived when a generation commits, never set by a proposal.
72    #[serde(default)]
73    pub security_floor: Option<GenerationId>,
74    pub units: BTreeMap<UnitId, (Observed, Vec<Evidence>)>,
75    pub overrides: Vec<Override>,
76    /// Atomic conditions that currently hold.
77    pub conditions: BTreeSet<Condition>,
78    /// Excluded from the digest.
79    pub resources: Resources,
80    pub core_hash: Hash,
81    /// §12d: the units whose traffic is held now: the journal's holds and
82    /// the host's (`/run/planarian/hold`), their union (fail closed). In the
83    /// digest: a hold changes what a transition does.
84    #[serde(default)]
85    pub holds: BTreeSet<UnitId>,
86    /// The reference time for staleness. Excluded from the digest.
87    pub observed_at: Timestamp,
88}
89
90impl SystemState {
91    /// The override holding `id` right now, if any.
92    pub fn override_of(&self, id: &UnitId) -> Option<&Override> {
93        self.overrides.iter().find(|s| &s.unit == id && s.holds_at(self.observed_at))
94    }
95
96    pub fn overridden(&self, id: &UnitId) -> bool {
97        self.override_of(id).is_some()
98    }
99
100    /// Whether the unit is expected to be in its desired status right now:
101    /// it is defined, its desired condition holds, and no override holds it.
102    pub fn expected(&self, id: &UnitId, defs: &UnitDefs) -> bool {
103        match defs.get(id) {
104            Some(def) => !self.overridden(id) && def.desired.when.holds(&self.conditions),
105            None => false,
106        }
107    }
108
109    /// Readiness probes of `id` whose fresh evidence is a failure. For a
110    /// device this is what `status` leaves out: §8 derives a bound device's
111    /// status from `Observed` alone, so one bound to a driver that cannot
112    /// drive it still reads Bound.
113    pub fn failing_readiness(&self, id: &UnitId, defs: &UnitDefs) -> Vec<ProbeId> {
114        let (Some(def), Some((_, evidence))) = (defs.get(id), self.units.get(id)) else { return vec![] };
115        def.probes
116            .iter()
117            .filter(|p| p.role == ProbeRole::Readiness)
118            .filter(|p| {
119                evidence
120                    .iter()
121                    .filter(|e| e.probe == p.id)
122                    .max_by_key(|e| e.at)
123                    .is_some_and(|e| e.fresh_at(self.observed_at) && matches!(e.result, ProbeResult::Fail(_)))
124            })
125            .map(|p| p.id.clone())
126            .collect()
127    }
128
129    /// The applicable Readiness probes whose latest evidence is a fresh pass
130    /// (§9: what a change may not take away, whatever the unit's status).
131    pub fn passing_readiness(&self, id: &UnitId, defs: &UnitDefs) -> Vec<ProbeId> {
132        let (Some(def), Some((_, evidence))) = (defs.get(id), self.units.get(id)) else { return vec![] };
133        def.probes
134            .iter()
135            .filter(|p| p.role == ProbeRole::Readiness)
136            .filter(|p| {
137                evidence
138                    .iter()
139                    .filter(|e| e.probe == p.id)
140                    .max_by_key(|e| e.at)
141                    .is_some_and(|e| e.fresh_at(self.observed_at) && matches!(e.result, ProbeResult::Pass))
142            })
143            .map(|p| p.id.clone())
144            .collect()
145    }
146
147    /// Status derivation, in the order §8 gives:
148    /// Overridden; not expected; definite Failed; Unknown on stale or timed-out
149    /// Readiness evidence; Running / Degraded from Active; otherwise from `Observed`.
150    /// Diagnostic probes never affect the result.
151    pub fn status(&self, id: &UnitId, defs: &UnitDefs) -> UnitStatus {
152        if let Some(o) = self.override_of(id) {
153            return UnitStatus::Overridden(o.desired);
154        }
155        let Some(def) = defs.get(id) else {
156            return UnitStatus::NotExpected;
157        };
158        if !def.desired.when.holds(&self.conditions) {
159            return UnitStatus::NotExpected;
160        }
161        let Some((observed, evidence)) = self.units.get(id) else {
162            return UnitStatus::Unknown;
163        };
164        if let Observed::Failed(r) = observed {
165            return UnitStatus::Failed(r.clone());
166        }
167
168        let mut failed = Vec::new();
169        for probe in def.probes.iter().filter(|p| p.role == ProbeRole::Readiness) {
170            let latest = evidence.iter().filter(|e| e.probe == probe.id).max_by_key(|e| e.at);
171            match latest {
172                Some(e) if e.fresh_at(self.observed_at) => match &e.result {
173                    ProbeResult::Pass => {}
174                    ProbeResult::Fail(_) => failed.push(probe.id.clone()),
175                    ProbeResult::Timeout | ProbeResult::Abandoned | ProbeResult::Error(_) => {
176                        return UnitStatus::Unknown;
177                    }
178                },
179                _ => return UnitStatus::Unknown,
180            }
181        }
182
183        match observed {
184            Observed::Active if failed.is_empty() => UnitStatus::Running,
185            Observed::Active => UnitStatus::Degraded(failed),
186            Observed::Unknown { .. } => UnitStatus::Unknown,
187            Observed::Transitioning { .. } => UnitStatus::Transitioning,
188            Observed::Absent => UnitStatus::Absent,
189            Observed::Detected => UnitStatus::Detected,
190            // systemd `inactive` maps to Bound; for anything but a device that is
191            // a stopped unit once desired state is considered (§8 systemd mapping).
192            Observed::Bound if def.id.kind == UnitKind::Device => UnitStatus::Bound,
193            Observed::Bound => UnitStatus::Stopped,
194            Observed::Failed(_) => unreachable!("handled above"),
195        }
196    }
197
198    /// Hash of everything that describes the system's state, excluding
199    /// `resources` and all observation timestamps (`observed_at`, `Evidence::at`,
200    /// `Unknown::since`, `Transitioning::since`). Two observations of an unchanged
201    /// system therefore have the same digest. Override deadlines (`Until::At`)
202    /// are desired-state data and are included.
203    pub fn digest(&self) -> Hash {
204        let view = DigestView {
205            generation: self.generation,
206            known_good: self.known_good,
207            live_tested: self.live_tested,
208            accepted_baseline: self.accepted_baseline,
209            units: self
210                .units
211                .iter()
212                .map(|(id, (obs, ev))| {
213                    let mut results: Vec<(&ProbeId, &ProbeResult)> = ev.iter().map(|e| (&e.probe, &e.result)).collect();
214                    results.sort_by(|a, b| a.0.cmp(b.0));
215                    (id, ObservedView::of(obs), results)
216                })
217                .collect(),
218            overrides: {
219                let mut s: Vec<&Override> = self.overrides.iter().collect();
220                s.sort();
221                s
222            },
223            conditions: &self.conditions,
224            core_hash: self.core_hash,
225            holds: &self.holds,
226        };
227        Hash::of(&serde_json::to_vec(&view).expect("digest view serializes"))
228    }
229}
230
231type UnitView<'a> = (&'a UnitId, ObservedView<'a>, Vec<(&'a ProbeId, &'a ProbeResult)>);
232
233#[derive(Serialize)]
234struct DigestView<'a> {
235    generation: GenerationId,
236    known_good: GenerationId,
237    live_tested: Option<GenerationId>,
238    accepted_baseline: Option<GenerationId>,
239    units: Vec<UnitView<'a>>,
240    overrides: Vec<&'a Override>,
241    conditions: &'a BTreeSet<Condition>,
242    core_hash: Hash,
243    holds: &'a BTreeSet<UnitId>,
244}
245
246#[derive(Serialize, PartialEq)]
247enum ObservedView<'a> {
248    Unknown { reason: &'a str },
249    Transitioning { from: Box<ObservedView<'a>> },
250    Absent,
251    Detected,
252    Bound,
253    Active,
254    Failed(&'a Reason),
255}
256
257impl<'a> ObservedView<'a> {
258    fn of(o: &'a Observed) -> ObservedView<'a> {
259        match o {
260            Observed::Unknown { reason, .. } => ObservedView::Unknown { reason },
261            Observed::Transitioning { from, .. } => {
262                ObservedView::Transitioning { from: Box::new(ObservedView::of(from)) }
263            }
264            Observed::Absent => ObservedView::Absent,
265            Observed::Detected => ObservedView::Detected,
266            Observed::Bound => ObservedView::Bound,
267            Observed::Active => ObservedView::Active,
268            Observed::Failed(r) => ObservedView::Failed(r),
269        }
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276    use crate::ids::Millis;
277    use crate::unit::*;
278
279    fn probe(id: &str, role: ProbeRole) -> ProbeDef {
280        ProbeDef {
281            id: ProbeId(id.into()),
282            kind: ProbeKind::SystemdActive,
283            timeout: Millis::secs(10),
284            origin: ProbeOrigin::Canonical,
285            role,
286            cadence: if role == ProbeRole::Readiness { Cadence::EveryTick } else { Cadence::OnDemand },
287            cost: ProbeCost::Trivial,
288            side_effects: if role == ProbeRole::Readiness { SideEffects::None } else { SideEffects::Audible },
289            establishes: String::new(),
290        }
291    }
292
293    fn def(id: &UnitId, when: Condition, probes: Vec<ProbeDef>) -> UnitDef {
294        UnitDef {
295            id: id.clone(),
296            desired: Desired { status: DesiredStatus::Running, when },
297            edges: vec![],
298            sensitivity: Sensitivities::default(),
299            restart_disruption: Disruption::Brief,
300            probes,
301            synonyms: vec![],
302            systemd_restart: None,
303            startup_grace: None,
304            runtime_repair: None,
305            stops_are_faults: false,
306            holds: false,
307            hold_allow: vec![],
308            hold_interfaces: vec![],
309        }
310    }
311
312    fn ev(id: &str, result: ProbeResult, at: u64) -> Evidence {
313        Evidence { probe: ProbeId(id.into()), result, at: Timestamp(at), stale_after: Millis(30_000) }
314    }
315
316    fn state(now: u64) -> SystemState {
317        SystemState {
318            generation: GenerationId(2),
319            known_good: GenerationId(1),
320            live_tested: None,
321            accepted_baseline: None,
322            security_floor: None,
323            units: BTreeMap::new(),
324            overrides: vec![],
325            conditions: BTreeSet::new(),
326            resources: Resources::default(),
327            core_hash: Hash::ZERO,
328            holds: BTreeSet::new(),
329            observed_at: Timestamp(now),
330        }
331    }
332
333    fn web() -> (UnitId, UnitDefs) {
334        let id = UnitId::service("gym-web");
335        let d = def(
336            &id,
337            Condition::Always,
338            vec![probe("http", ProbeRole::Readiness), probe("tone", ProbeRole::Diagnostic)],
339        );
340        (id.clone(), [(id, d)].into())
341    }
342
343    #[test]
344    fn running_needs_fresh_readiness_only() {
345        let (id, defs) = web();
346        let mut s = state(100_000);
347        // Diagnostic evidence is long stale; it must not matter (catalog: probe-diagnostic-stale-healthy).
348        s.units.insert(
349            id.clone(),
350            (Observed::Active, vec![ev("http", ProbeResult::Pass, 90_000), ev("tone", ProbeResult::Pass, 1)]),
351        );
352        assert_eq!(s.status(&id, &defs), UnitStatus::Running);
353        s.observed_at = Timestamp(200_000);
354        assert_eq!(s.status(&id, &defs), UnitStatus::Unknown);
355    }
356
357    #[test]
358    fn failed_beats_stale() {
359        // Catalog: probe-failed-beats-stale.
360        let (id, defs) = web();
361        let mut s = state(1_000_000);
362        s.units.insert(id.clone(), (Observed::Failed(Reason::ExitCode(137)), vec![ev("http", ProbeResult::Pass, 1)]));
363        assert_eq!(s.status(&id, &defs), UnitStatus::Failed(Reason::ExitCode(137)));
364    }
365
366    #[test]
367    fn timeout_is_unknown_not_failed() {
368        let (id, defs) = web();
369        let mut s = state(100_000);
370        s.units.insert(id.clone(), (Observed::Active, vec![ev("http", ProbeResult::Timeout, 99_000)]));
371        assert_eq!(s.status(&id, &defs), UnitStatus::Unknown);
372    }
373
374    #[test]
375    fn degraded_and_stopped() {
376        let (id, defs) = web();
377        let mut s = state(100_000);
378        s.units.insert(id.clone(), (Observed::Active, vec![ev("http", ProbeResult::Fail("503".into()), 99_000)]));
379        assert_eq!(s.status(&id, &defs), UnitStatus::Degraded(vec![ProbeId("http".into())]));
380        s.units.insert(id.clone(), (Observed::Bound, vec![ev("http", ProbeResult::Fail("refused".into()), 99_000)]));
381        assert_eq!(s.status(&id, &defs), UnitStatus::Stopped);
382    }
383
384    #[test]
385    fn suppressed_and_not_expected() {
386        let id = UnitId::target("graphical-session");
387        let defs: UnitDefs =
388            [(id.clone(), def(&id, Condition::SessionEstablished { uid: None, seated: false }, vec![]))].into();
389        let mut s = state(100_000);
390        s.units.insert(id.clone(), (Observed::Bound, vec![]));
391        // Idle greeter: nobody logged in (catalog: login-greeter-idle).
392        assert_eq!(s.status(&id, &defs), UnitStatus::NotExpected);
393        assert!(!s.expected(&id, &defs));
394        s.conditions.insert(Condition::SessionEstablished { uid: Some(1000), seated: false });
395        assert_eq!(s.status(&id, &defs), UnitStatus::Stopped);
396        assert!(s.expected(&id, &defs));
397        s.overrides.push(Override {
398            unit: id.clone(),
399            until: Until::At(Timestamp(200_000)),
400            reason: String::new(),
401            by: Actor::user(1000),
402            desired: OverrideDesired::LeaveAlone,
403            attribution: Attribution::Attested,
404            acknowledged: false,
405        });
406        assert_eq!(s.status(&id, &defs), UnitStatus::Overridden(OverrideDesired::LeaveAlone));
407        s.observed_at = Timestamp(200_000);
408        assert_eq!(s.status(&id, &defs), UnitStatus::Stopped);
409    }
410
411    #[test]
412    fn digest_ignores_timestamps_and_resources() {
413        let (id, _) = web();
414        let mut a = state(100_000);
415        a.units.insert(id.clone(), (Observed::Active, vec![ev("http", ProbeResult::Pass, 99_000)]));
416        let mut b = a.clone();
417        b.observed_at = Timestamp(555_555);
418        b.resources.mem_available_mb = 7;
419        b.units.insert(id.clone(), (Observed::Active, vec![ev("http", ProbeResult::Pass, 1)]));
420        assert_eq!(a.digest(), b.digest());
421        b.units.insert(id.clone(), (Observed::Active, vec![ev("http", ProbeResult::Fail("x".into()), 1)]));
422        assert_ne!(a.digest(), b.digest());
423        let mut c = a.clone();
424        c.generation = GenerationId(3);
425        assert_ne!(a.digest(), c.digest());
426    }
427
428    /// §18.4, over random states: perturbing resources or any observation
429    /// timestamp never changes the digest; changing what the state says
430    /// always does.
431    mod digest_props {
432        use super::*;
433        use proptest::prelude::*;
434        use proptest::test_runner::{Config, FailurePersistence};
435
436        fn observed() -> impl Strategy<Value = Observed> {
437            prop_oneof![
438                Just(Observed::Active),
439                Just(Observed::Bound),
440                Just(Observed::Absent),
441                (0..3i32).prop_map(|c| Observed::Failed(Reason::ExitCode(c))),
442                "[a-c]{1,3}".prop_map(|r| Observed::Unknown { since: Timestamp(1), reason: r }),
443                Just(Observed::Transitioning { from: Box::new(Observed::Active), since: Timestamp(1) }),
444            ]
445        }
446
447        fn result() -> impl Strategy<Value = ProbeResult> {
448            prop_oneof![
449                Just(ProbeResult::Pass),
450                Just(ProbeResult::Timeout),
451                Just(ProbeResult::Abandoned),
452                "[xy]{1,2}".prop_map(ProbeResult::Fail)
453            ]
454        }
455
456        prop_compose! {
457            fn system()(
458                generation in 1..4u64,
459                units in proptest::collection::btree_map(
460                    "[a-d]{1,2}",
461                    (observed(), proptest::collection::vec(("[pq]", result()), 0..3)),
462                    0..4,
463                ),
464            ) -> SystemState {
465                let mut s = state(1_000);
466                s.generation = GenerationId(generation);
467                for (name, (o, ev)) in units {
468                    let ev = ev.into_iter().map(|(p, r)| super::ev(&p, r, 900)).collect();
469                    s.units.insert(UnitId::service(&name), (o, ev));
470                }
471                s
472            }
473        }
474
475        /// Every timestamp and resource figure in the state, moved.
476        fn perturb(s: &SystemState, by: u64) -> SystemState {
477            let mut t = s.clone();
478            t.observed_at = Timestamp(t.observed_at.0 + by);
479            t.resources.mem_available_mb += by;
480            t.resources.mem_total_mb += 2 * by;
481            for (o, ev) in t.units.values_mut() {
482                match o {
483                    Observed::Unknown { since, .. } | Observed::Transitioning { since, .. } => since.0 += by,
484                    _ => {}
485                }
486                for e in ev {
487                    e.at.0 += by;
488                }
489            }
490            t
491        }
492
493        proptest! {
494            #![proptest_config(Config { cases: 256, failure_persistence: None::<Box<dyn FailurePersistence>>, ..Config::default() })]
495
496            #[test]
497            fn timestamps_and_resources_do_not_count(s in system(), by in 1..1_000_000u64) {
498                prop_assert_eq!(s.digest(), perturb(&s, by).digest());
499            }
500
501            #[test]
502            fn what_the_state_says_does(s in system(), o in observed()) {
503                let mut t = s.clone();
504                t.generation = GenerationId(s.generation.0 + 10);
505                prop_assert_ne!(s.digest(), t.digest());
506                let mut u = s.clone();
507                u.overrides.push(Override {
508                    unit: UnitId::service("z"),
509                    until: Until::Explicit,
510                    reason: String::new(),
511                    by: Actor::Daemon,
512                    desired: OverrideDesired::LeaveAlone,
513                    attribution: Attribution::Attested,
514                    acknowledged: false,
515                });
516                prop_assert_ne!(s.digest(), u.digest());
517                if let Some((id, (old, ev))) = s.units.iter().next() {
518                    let same = ObservedView::of(old) == ObservedView::of(&o);
519                    let mut v = s.clone();
520                    v.units.insert(id.clone(), (o, ev.clone()));
521                    prop_assert_eq!(s.digest() == v.digest(), same);
522                }
523            }
524        }
525    }
526}
planarian_schema/state.rs

planarian_schema/
state.rs
