planarian_state/
lib.rs

1//! State assembly (SPEC §10): each tick, observe every unit, run the probes
2//! that are due, evaluate conditions, and assemble a `SystemState`.
3//!
4//! Desired state comes from the evaluated manifest, observed state from the
5//! backends, health from the probes. The three stay separate: `SystemState`
6//! keeps observations and evidence, and status is derived on demand.
7//!
8//! Canonical probes come from the installed core package, never from the
9//! generation's manifest (I10). The manifest can add `User` probes; a
10//! definition it gives for a canonical probe id is replaced by the core's.
11//!
12//! No process spawning and no reads of `/proc`, `/sys` or `/nix`: everything
13//! goes through the backend traits (§6).
14
15use planarian_schema::{
16    Backends, Cadence, Clock, Condition, Disruption, DriverRef, Evidence, GenerationBackend, GenerationId, Hash,
17    InventoryBackend, Millis, Observed, Override, PowerSource, ProbeBackend, ProbeCost, ProbeDef, ProbeId, ProbeKind,
18    ProbeOrigin, ProbeResult, ProbeRole, Reason, Scope, SessionBackend, SessionInfo, SystemState, Timestamp,
19    UnitBackend, UnitDef, UnitDefs, UnitId, UnitKind,
20};
21use std::collections::{BTreeMap, BTreeSet};
22
23/// Quiet mode's probe interval for units foreground work does not depend on.
24pub const QUIET_INTERVAL: Millis = Millis::secs(60);
25
26pub const TICK: Millis = Millis::secs(5);
27pub const MIN_STALE: Millis = Millis::secs(30);
28pub const HARDWARE_CADENCE: Millis = Millis::secs(60);
29pub const HARDWARE_CADENCE_BATTERY: Millis = Millis::secs(300);
30/// Warn if a tick takes longer than this (§10).
31pub const TICK_BUDGET: Millis = Millis(500);
32
33pub use planarian_schema::Manifest;
34
35/// Canonical probes shipped in the core package, by unit name.
36pub type Canonical = BTreeMap<String, Vec<ProbeDef>>;
37
38/// Replace or add the core's canonical probes on a unit (I10). Probes the
39/// manifest marks Canonical but the core does not ship are demoted to User.
40pub fn apply_canonical(def: &mut UnitDef, canonical: &Canonical) {
41    let core = canonical.get(&def.id.name).cloned().unwrap_or_default();
42    let core_ids: BTreeSet<&ProbeId> = core.iter().map(|p| &p.id).collect();
43    def.probes.retain(|p| !core_ids.contains(&p.id));
44    for p in def.probes.iter_mut() {
45        p.origin = ProbeOrigin::User;
46    }
47    def.probes.extend(core);
48}
49
50/// Baselines and overrides from the journal, which `state` does not own.
51#[derive(Clone, Debug, Default)]
52pub struct Context {
53    pub known_good: GenerationId,
54    pub live_tested: Option<GenerationId>,
55    pub accepted_baseline: Option<GenerationId>,
56    /// §12e, from the ledger.
57    pub security_floor: Option<GenerationId>,
58    pub overrides: Vec<Override>,
59    pub core_hash: Hash,
60}
61
62#[derive(Clone, Debug, Default)]
63pub struct Assembler {
64    /// Quiet mode (§12f): Readiness probes of units foreground work does not
65    /// depend on drop to their slowest interval.
66    pub quiet: bool,
67    manifest: Manifest,
68    canonical: Canonical,
69    evidence: BTreeMap<(UnitId, ProbeId), Evidence>,
70    last_run: BTreeMap<(UnitId, ProbeId), Timestamp>,
71    on_demand: BTreeSet<(UnitId, ProbeId)>,
72    pub failed_since: BTreeMap<UnitId, Timestamp>,
73    pub known_drivers: BTreeMap<UnitId, DriverRef>,
74    pub sessions: Vec<SessionInfo>,
75    /// When each session id was first observed (stands in for an unknown creation time).
76    session_seen: BTreeMap<String, Timestamp>,
77    pub on_battery: bool,
78    /// Ticks that went over budget (logged by the daemon).
79    pub slow_ticks: u64,
80    /// The last tick over budget, if the daemon has not reported it yet:
81    /// how long it took and the probes that ran in it (P9).
82    pub slow_tick: Option<(Millis, Vec<String>)>,
83}
84
85impl Assembler {
86    pub fn new(manifest: Manifest, canonical: Canonical) -> Assembler {
87        let mut a = Assembler { canonical, ..Default::default() };
88        a.set_manifest(manifest);
89        a
90    }
91
92    /// A new generation's manifest (after a commit, revert or boot).
93    pub fn set_manifest(&mut self, mut manifest: Manifest) {
94        for def in manifest.units.values_mut() {
95            apply_canonical(def, &self.canonical);
96        }
97        for def in manifest.user_units.iter_mut() {
98            apply_canonical(def, &self.canonical);
99        }
100        self.manifest = manifest;
101    }
102
103    pub fn manifest(&self) -> &Manifest {
104        &self.manifest
105    }
106
107    /// System units plus one instance of each user template per user with an
108    /// established session (§10: a user service is desired only while that
109    /// user's session exists).
110    pub fn defs(&self) -> UnitDefs {
111        let mut defs = self.manifest.units.clone();
112        let uids: BTreeSet<u32> = self.sessions.iter().filter(|s| s.establishes()).map(|s| s.uid).collect();
113        // §9: a session-bound unit (graphical-session) is one instance per
114        // established session, so a second session's failure is its own incident.
115        let bound: Vec<UnitDef> = defs.values().filter(|d| session_bound(d)).cloned().collect();
116        for t in bound {
117            defs.remove(&t.id);
118            // S05: a template that asks for a seat has no instance for an
119            // SSH-only session.
120            let seated = t.desired.when.needs_seat();
121            for s in self.sessions.iter().filter(|s| s.establishes() && (s.seat.is_some() || !seated)) {
122                let id = UnitId {
123                    kind: t.id.kind,
124                    name: t.id.name.clone(),
125                    scope: Scope::User { uid: s.uid, session: Some(s.id.clone()) },
126                };
127                let mut d = t.clone();
128                d.id = id.clone();
129                d.desired.when = Condition::All(vec![
130                    t.desired.when.clone(),
131                    Condition::SessionEstablished { uid: Some(s.uid), seated },
132                ]);
133                for e in d.edges.iter_mut() {
134                    e.from = id.clone();
135                }
136                defs.insert(id, d);
137            }
138        }
139        for t in &self.manifest.user_units {
140            for uid in &uids {
141                let id = UnitId {
142                    kind: t.id.kind,
143                    name: t.id.name.clone(),
144                    scope: Scope::User { uid: *uid, session: None },
145                };
146                let mut d = t.clone();
147                d.id = id.clone();
148                d.desired.when = Condition::All(vec![
149                    t.desired.when.clone(),
150                    Condition::SessionEstablished { uid: Some(*uid), seated: false },
151                ]);
152                for e in d.edges.iter_mut() {
153                    e.from = id.clone();
154                    if let Scope::User { .. } = e.to.scope {
155                        e.to.scope = Scope::User { uid: *uid, session: None };
156                    }
157                }
158                defs.insert(id, d);
159            }
160        }
161        defs
162    }
163
164    /// Run a Diagnostic (or any) probe on the next tick: user-initiated diagnostics (§10).
165    pub fn request(&mut self, unit: &UnitId, probe: &ProbeId) {
166        self.on_demand.insert((unit.clone(), probe.clone()));
167    }
168
169    /// Re-run every Readiness probe of `unit` now (diagnostic incident: re-observe).
170    pub fn reprobe(&mut self, unit: &UnitId) {
171        self.last_run.retain(|(u, _), _| u != unit);
172    }
173
174    fn interval(&self, p: &ProbeDef) -> Option<Millis> {
175        let base = match p.cadence {
176            Cadence::EveryTick | Cadence::OnEvent => TICK,
177            Cadence::Every(d) => d,
178            Cadence::OnDemand => return None,
179        };
180        let base = if p.cost == ProbeCost::WakesHardware {
181            let floor = if self.on_battery { HARDWARE_CADENCE_BATTERY } else { HARDWARE_CADENCE };
182            Millis(base.0.max(floor.0))
183        } else {
184            base
185        };
186        // §10: "`per` enforced by the cadence scheduler". A Measurement probe's
187        // cadence is never faster than its declared rate. Runs on request
188        // (`request`, `reprobe`) are not held back by it.
189        Some(Millis(base.0.max(per(p).map_or(0, |m| m.0))))
190    }
191
192    fn stale_after(&self, p: &ProbeDef) -> Millis {
193        let interval = self.interval(p).unwrap_or(TICK);
194        Millis((interval.0 * 3).max(p.timeout.0 * 2 + TICK.0).max(MIN_STALE.0))
195    }
196
197    /// Quiet mode holds back a probe of a unit foreground work does not
198    /// plausibly depend on (audio, the compositor, input devices, the
199    /// network), to once a minute.
200    fn quieted(&self, def: &UnitDef, p: &ProbeDef, now: Timestamp) -> bool {
201        if !self.quiet || self.on_demand.contains(&(def.id.clone(), p.id.clone())) {
202            return false;
203        }
204        let depended_on = matches!(def.id.scope, Scope::User { .. })
205            || def.id.kind == UnitKind::Device
206            || matches!(def.restart_disruption, Disruption::SessionLoss | Disruption::ConnectivityLoss);
207        !depended_on
208            && self.last_run.get(&(def.id.clone(), p.id.clone())).is_some_and(|last| now.0 < last.0 + QUIET_INTERVAL.0)
209    }
210
211    fn due(&self, unit: &UnitId, p: &ProbeDef, now: Timestamp) -> bool {
212        let key = (unit.clone(), p.id.clone());
213        if self.on_demand.contains(&key) {
214            return true;
215        }
216        // Side-effecting probes only ever run on demand (§10, §18.4); Diagnostic
217        // probes never run on a cadence.
218        if p.role == ProbeRole::Diagnostic || p.check_contract().is_err() {
219            return false;
220        }
221        match (self.interval(p), self.last_run.get(&key)) {
222            (None, _) => false,
223            (Some(_), None) => true,
224            (Some(i), Some(last)) => now.0 >= last.0 + i.0,
225        }
226    }
227
228    fn observe_unit<B: Backends>(
229        &mut self,
230        b: &B,
231        def: &UnitDef,
232        devices: &BTreeMap<UnitId, Option<DriverRef>>,
233    ) -> Observed {
234        let id = &def.id;
235        let now = b.clock().now();
236        if id.kind == UnitKind::Device {
237            if id.name.starts_with("fs-") {
238                return Observed::Active; // a mounted filesystem; its probe judges space
239            }
240            return match devices.get(id) {
241                None => Observed::Absent,
242                Some(None) => Observed::Detected,
243                Some(Some(d)) => {
244                    self.known_drivers.insert(id.clone(), d.clone());
245                    Observed::Bound
246                }
247            };
248        }
249        if id.name.starts_with("external:") {
250            return Observed::Active; // nothing local to observe; its probes decide
251        }
252        if let Some((target, within)) = def.probes.iter().find_map(|p| match &p.kind {
253            ProbeKind::SessionTarget { target, within } => Some((target.clone(), *within)),
254            _ => None,
255        }) {
256            return match self.session_of(id) {
257                Some(s) => self.observe_session_target(b, &target, within, now, s),
258                None => Observed::Absent,
259            };
260        }
261        match b.units().observe(id) {
262            Ok(o) => o,
263            Err(e) => Observed::Unknown { since: now, reason: format!("backend: {e:?}") },
264        }
265    }
266
267    /// The logind session a session-bound unit instance belongs to.
268    fn session_of(&self, id: &UnitId) -> Option<&SessionInfo> {
269        match &id.scope {
270            Scope::User { uid, session: Some(sid) } => {
271                self.sessions.iter().find(|s| &s.id == sid && s.uid == *uid && s.establishes())
272            }
273            _ => None,
274        }
275    }
276
277    /// Whether this session's desktop is up (§9): the compositor in the
278    /// session's own cgroup is running, when the backend can see that; else
279    /// the target in the user's manager, which is only corroboration (for a
280    /// second session of the same uid it was reached already).
281    fn session_up<B: Backends>(&self, b: &B, target: &str, s: &SessionInfo) -> bool {
282        if let Some(up) = b.sessions().compositor_running(s) {
283            return up;
284        }
285        let tid = UnitId {
286            kind: UnitKind::Target,
287            name: target.into(),
288            scope: Scope::User { uid: s.uid, session: Some(s.id.clone()) },
289        };
290        matches!(b.units().observe(&tid), Ok(Observed::Active))
291    }
292
293    /// §9 graphical-session, for one session: up within `within` of the
294    /// session's creation. Before the deadline it is Transitioning; a missed
295    /// deadline makes it inactive, and its probe fails.
296    fn observe_session_target<B: Backends>(
297        &self,
298        b: &B,
299        target: &str,
300        within: Millis,
301        now: Timestamp,
302        s: &SessionInfo,
303    ) -> Observed {
304        if self.session_up(b, target, s) {
305            Observed::Active
306        } else if now.0 < s.created.0 + within.0 {
307            Observed::Transitioning { from: Box::new(Observed::Bound), since: now }
308        } else {
309            Observed::Bound
310        }
311    }
312
313    fn session_target_result<B: Backends>(
314        &self,
315        b: &B,
316        id: &UnitId,
317        target: &str,
318        within: Millis,
319        now: Timestamp,
320    ) -> ProbeResult {
321        let Some(s) = self.session_of(id) else { return ProbeResult::Error("no such session".into()) };
322        if self.session_up(b, target, s) || now.0 < s.created.0 + within.0 {
323            ProbeResult::Pass
324        } else {
325            ProbeResult::Fail(format!(
326                "the desktop did not come up within {}s of session {} (uid {})",
327                within.0 / 1000,
328                s.id,
329                s.uid
330            ))
331        }
332    }
333
334    /// One tick: observe, probe, assemble.
335    pub fn observe<B: Backends>(&mut self, b: &B, ctx: &Context) -> SystemState {
336        let started = b.clock().now();
337        let now = started;
338
339        self.sessions = b.sessions().sessions().unwrap_or_default();
340        for s in self.sessions.iter_mut() {
341            let seen = *self.session_seen.entry(s.id.clone()).or_insert(now);
342            if s.created.0 == 0 {
343                s.created = seen;
344            }
345        }
346        let live: BTreeSet<String> = self.sessions.iter().map(|s| s.id.clone()).collect();
347        self.session_seen.retain(|id, _| live.contains(id));
348        let power = b.sessions().power();
349        self.on_battery = power == Some(PowerSource::Battery);
350        let devices: BTreeMap<UnitId, Option<DriverRef>> =
351            b.inventory().devices().unwrap_or_default().into_iter().map(|d| (d.id, d.driver)).collect();
352
353        let mut conditions = BTreeSet::new();
354        for s in self.sessions.iter().filter(|s| s.establishes()) {
355            conditions.insert(Condition::SessionEstablished { uid: Some(s.uid), seated: s.seat.is_some() });
356        }
357        // §10 quiet mode: foreground work in a live session.
358        for sid in b.sessions().foreground() {
359            if self.sessions.iter().any(|s| s.id == sid && s.establishes()) {
360                conditions.insert(Condition::Foreground { session: sid });
361            }
362        }
363        match power {
364            Some(PowerSource::Ac) => {
365                conditions.insert(Condition::OnAcPower);
366            }
367            Some(PowerSource::Battery) => {
368                conditions.insert(Condition::OnBattery);
369            }
370            None => {}
371        }
372        for d in devices.keys() {
373            conditions.insert(Condition::DeviceConnected(d.clone()));
374        }
375
376        let defs = self.defs();
377        let mut observed: BTreeMap<UnitId, Observed> = BTreeMap::new();
378        for def in defs.values() {
379            let o = self.observe_unit(b, def, &devices);
380            match &o {
381                Observed::Failed(_) => {
382                    self.failed_since.entry(def.id.clone()).or_insert(now);
383                }
384                _ => {
385                    self.failed_since.remove(&def.id);
386                }
387            }
388            observed.insert(def.id.clone(), o);
389        }
390
391        // Probes that are due. SystemdActive and SessionTarget are answered here;
392        // the rest go to the probe backend as one concurrent batch.
393        let mut batch: Vec<(UnitId, ProbeDef)> = vec![];
394        let mut local: Vec<(UnitId, ProbeDef, ProbeResult)> = vec![];
395        for def in defs.values() {
396            for p in &def.probes {
397                if self.quieted(def, p, now) {
398                    // Held back, not lost: its evidence stays valid for the
399                    // slower interval.
400                    if let Some(e) = self.evidence.get_mut(&(def.id.clone(), p.id.clone())) {
401                        e.stale_after = Millis(e.stale_after.0.max(QUIET_INTERVAL.0 * 3));
402                    }
403                    continue;
404                }
405                if !self.due(&def.id, p, now) {
406                    continue;
407                }
408                match &p.kind {
409                    ProbeKind::SystemdActive => {
410                        let r = match &observed[&def.id] {
411                            Observed::Active | Observed::Bound if def.id.kind == UnitKind::Device => ProbeResult::Pass,
412                            Observed::Active => ProbeResult::Pass,
413                            Observed::Unknown { reason, .. } => ProbeResult::Error(reason.clone()),
414                            other => ProbeResult::Fail(format!("{other:?}")),
415                        };
416                        local.push((def.id.clone(), p.clone(), r));
417                    }
418                    ProbeKind::SessionTarget { target, within } => {
419                        let r = self.session_target_result(b, &def.id, target, *within, now);
420                        local.push((def.id.clone(), p.clone(), r));
421                    }
422                    _ => batch.push((def.id.clone(), p.clone())),
423                }
424            }
425        }
426        let results = if batch.is_empty() { vec![] } else { b.probes().run_batch(&batch) };
427        for ((unit, p), r) in batch.into_iter().zip(results) {
428            local.push((unit, p, r));
429        }
430        let ran: Vec<String> = local.iter().map(|(u, p, _)| format!("{u}/{}", p.id.0)).collect();
431        for (unit, p, result) in local {
432            let key = (unit.clone(), p.id.clone());
433            self.on_demand.remove(&key);
434            self.last_run.insert(key.clone(), now);
435            let stale_after = self.stale_after(&p);
436            self.evidence.insert(key, Evidence { probe: p.id.clone(), result, at: now, stale_after });
437        }
438        // Drop evidence for units or probes that no longer exist.
439        self.evidence.retain(|(u, pid), _| defs.get(u).is_some_and(|d| d.probes.iter().any(|p| &p.id == pid)));
440
441        let units = observed
442            .into_iter()
443            .map(|(id, o)| {
444                let ev: Vec<Evidence> =
445                    self.evidence.iter().filter(|((u, _), _)| *u == id).map(|(_, e)| e.clone()).collect();
446                (id, (o, ev))
447            })
448            .collect();
449
450        let generation = b.generations().current().unwrap_or(ctx.known_good);
451        let resources = b.inventory().resources().unwrap_or_default();
452        let finished = b.clock().now();
453        let took = finished.0.saturating_sub(started.0);
454        if took > TICK_BUDGET.0 {
455            self.slow_ticks += 1;
456            self.slow_tick = Some((Millis(took), ran));
457        }
458        SystemState {
459            generation,
460            known_good: ctx.known_good,
461            live_tested: ctx.live_tested,
462            accepted_baseline: ctx.accepted_baseline,
463            security_floor: ctx.security_floor,
464            units,
465            overrides: ctx.overrides.clone(),
466            conditions,
467            resources,
468            core_hash: ctx.core_hash,
469            // §12d: filled by the daemon from its journal and the host.
470            holds: Default::default(),
471            observed_at: now,
472        }
473    }
474}
475
476/// A Measurement probe's declared rate (§10): its cadence runs at most once
477/// per `per`.
478fn per(p: &ProbeDef) -> Option<Millis> {
479    match p.side_effects {
480        planarian_schema::SideEffects::Measurement { per, .. } if per.0 > 0 => Some(per),
481        _ => None,
482    }
483}
484
485/// A `Reason` for a unit observed failed with a backend error.
486pub fn backend_reason(msg: &str) -> Reason {
487    Reason::Backend(msg.into())
488}
489
490#[cfg(test)]
491mod tests;
492
493/// A unit whose readiness is per logind session (§9 graphical-session).
494fn session_bound(d: &UnitDef) -> bool {
495    d.id.scope == Scope::System && d.probes.iter().any(|p| matches!(p.kind, ProbeKind::SessionTarget { .. }))
496}
planarian_state/lib.rs

planarian_state/
lib.rs
