planarian_exec/
host.rs

1//! The real `txn::Host`: the kernel's view of the backends, plus the probe
2//! check and re-observation the protocol needs (§9 executor-owned
3//! requirements, §12 step 7). The daemon and the watchdog both use it, so
4//! the watchdog can verify a candidate on its own from the journaled
5//! `CheckContext`.
6
7use planarian_schema::spec::DEFAULT_SETTLE;
8use planarian_schema::*;
9use planarian_state::{Assembler, Canonical, Context};
10use planarian_txn::kernel::{CheckPhase, Host, Verdict};
11use std::collections::BTreeSet;
12use std::sync::{Arc, Mutex};
13
14/// How long a check keeps re-observing before a non-Running unit counts as a failure.
15pub const CHECK_PATIENCE: Millis = Millis::secs(30);
16pub const CHECK_POLL: Millis = Millis::secs(2);
17
18/// Generations that garbage collection must keep, maintained by the executor (I3).
19pub type Protect = Arc<Mutex<Vec<GenerationId>>>;
20
21pub struct ExecHost<B: Backends> {
22    pub b: B,
23    pub canonical: Canonical,
24    pub protect: Protect,
25    /// Baselines for the probe runs' state assembly.
26    pub ctx: Arc<Mutex<Context>>,
27}
28
29impl<B: Backends + Clone> Clone for ExecHost<B> {
30    fn clone(&self) -> Self {
31        ExecHost {
32            b: self.b.clone(),
33            canonical: self.canonical.clone(),
34            protect: self.protect.clone(),
35            ctx: self.ctx.clone(),
36        }
37    }
38}
39
40impl<B: Backends> ExecHost<B> {
41    pub fn new(b: B, canonical: Canonical) -> Self {
42        ExecHost { b, canonical, protect: Arc::new(Mutex::new(vec![])), ctx: Arc::new(Mutex::new(Context::default())) }
43    }
44
45    fn ctx(&self) -> Context {
46        self.ctx.lock().unwrap_or_else(|e| e.into_inner()).clone()
47    }
48
49    fn settle(&self, settle: &Settle) {
50        match settle {
51            Settle::Duration(d) => self.b.clock().sleep(*d),
52            Settle::Ready { unit, timeout } => {
53                let deadline = self.b.clock().now().plus(*timeout);
54                while self.b.clock().now() < deadline {
55                    if matches!(self.b.units().observe(unit), Ok(Observed::Active)) {
56                        break;
57                    }
58                    self.b.clock().sleep(Millis::secs(1));
59                }
60            }
61        }
62    }
63
64    /// Observe once with a fresh assembler over `manifest` (every probe due).
65    fn observe(&self, manifest: &Manifest) -> (SystemState, UnitDefs) {
66        let mut a = Assembler::new(manifest.clone(), self.canonical.clone());
67        let s = a.observe(&self.b, &self.ctx());
68        (s, a.defs())
69    }
70
71    /// The requirements a check enforces; returns what failed.
72    fn violations(&self, txn: &Transaction, phase: CheckPhase, s: &SystemState, defs: &UnitDefs) -> Vec<String> {
73        let mut out = vec![];
74        let want_gen = match phase {
75            CheckPhase::Candidate => txn.candidate.as_ref().map(|(g, _)| *g),
76            CheckPhase::Revert => txn.candidate.as_ref().map(|_| txn.revert_to),
77        };
78        if let Some(g) = want_gen
79            && s.generation != g
80        {
81            out.push(format!("running generation is {} (expected {})", s.generation.0, g.0));
82        }
83        if phase == CheckPhase::Candidate {
84            for (u, want) in &txn.expected.units {
85                // A unit the candidate no longer defines can't be held to a status.
86                if !defs.contains_key(u) {
87                    continue;
88                }
89                let got = s.status(u, defs);
90                if &got != want {
91                    out.push(format!("{u}: expected {want:?}, observed {got:?}"));
92                } else if *want == UnitStatus::Bound {
93                    // Bound to a driver that cannot drive it (§12 DeviceOp
94                    // gym case): the device's own checks say so.
95                    let failing = s.failing_readiness(u, defs);
96                    if !failing.is_empty() {
97                        out.push(format!(
98                            "{u}: bound, but {} failing",
99                            failing.iter().map(|p| p.0.as_str()).collect::<Vec<_>>().join(", ")
100                        ));
101                    }
102                }
103            }
104        }
105        // Executor-owned (§9): every unit that is desired-present under these
106        // defs and was Running before is Running now. This covers both "no
107        // unit outside may_change leaves Running" and "no applicable Readiness
108        // probe that passed before fails after" (Running needs every applicable
109        // Readiness probe passing, I4).
110        for (u, before) in &txn.check.pre {
111            if *before != UnitStatus::Running || !defs.contains_key(u) || !s.expected(u, defs) {
112                continue;
113            }
114            let now = s.status(u, defs);
115            if now == UnitStatus::Running {
116                continue;
117            }
118            // A revert neither causes nor cures a full filesystem (§12c: /boot
119            // full at publish is a Checking failure and the change reverts);
120            // the storage units' own incidents take it from there. Anything
121            // else that did not come back fails the revert.
122            if phase == CheckPhase::Revert && matches!(now, UnitStatus::Degraded(_)) && space_only(u, s, defs) {
123                continue;
124            }
125            out.push(format!("{u} was Running before and is now {now:?}"));
126        }
127        // Per probe (§9, R07): a Readiness probe that passed before passes
128        // now, for every unit that is still expected under these defs, so a
129        // unit already Degraded by one probe cannot lose another unnoticed.
130        for (u, passing) in &txn.check.pre_passing {
131            if !defs.contains_key(u) || !s.expected(u, defs) {
132                continue;
133            }
134            let failing = s.failing_readiness(u, defs);
135            let lost: Vec<&str> = passing.iter().filter(|p| failing.contains(p)).map(|p| p.0.as_str()).collect();
136            if lost.is_empty() || (phase == CheckPhase::Revert && space_only(u, s, defs)) {
137                continue;
138            }
139            out.push(format!("{u}: {} passed before and fails now", lost.join(", ")));
140        }
141        out
142    }
143}
144
145impl<B: Backends> Host for ExecHost<B> {
146    type Gen = B::Gens;
147    type Inv = B::Inv;
148
149    fn floor_fixes(&self, txn: &Transaction, floor: Option<GenerationId>) -> Result<Vec<String>, Outcome> {
150        let gens = self.b.generations();
151        let Some((cand, _)) = txn.candidate.clone() else { return Ok(vec![]) };
152        let eval = |g: GenerationId| gens.path_of(g).and_then(|p| gens.evaluate_built(&p));
153        let c = eval(cand)?;
154        // The floor and every generation between it and the candidate (in A),
155        // so a Revert to an older generation is never a fix (§12e).
156        let mut earlier: Vec<GenerationId> = gens.boot_entries()?;
157        earlier.push(txn.revert_to);
158        earlier.sort();
159        earlier.dedup();
160        earlier.retain(|g| g.0 < cand.0 && floor.is_none_or(|f| g.0 >= f.0));
161        let earlier: Vec<EvaluatedGeneration> = earlier.into_iter().map(eval).collect::<Result<_, _>>()?;
162        Ok(c.packages
163            .keys()
164            .filter(|p| !c.insecure.contains(*p))
165            .filter(|p| earlier.iter().all(|e| e.insecure.contains(*p) || !e.packages.contains_key(*p)))
166            .filter(|p| earlier.iter().any(|e| e.insecure.contains(*p)))
167            .cloned()
168            .collect())
169    }
170
171    fn generations(&self) -> &B::Gens {
172        self.b.generations()
173    }
174
175    fn inventory(&self) -> &B::Inv {
176        self.b.inventory()
177    }
178
179    fn execute_runtime(&self, t: &Transition, op: &OpId) -> Outcome {
180        match t {
181            Transition::RestartUnit(u) => self.b.units().restart(u, op),
182            Transition::StopUnit(u) => self.b.units().stop(u, op),
183            Transition::CollectGarbage { emergency } => {
184                let protect = self.protect.lock().unwrap_or_else(|e| e.into_inner()).clone();
185                self.b.generations().collect_garbage(&protect, *emergency, op)
186            }
187            Transition::SetRuntimeOption { path, value } => self.b.runtime().set(path, value, op),
188            other => Outcome::Failed(format!("not a runtime operation: {other:?}")),
189        }
190    }
191
192    fn reobserve(&self, txn: &Transaction) -> Outcome {
193        self.settle(&txn.expected.settle);
194        let manifest = txn.check.candidate.clone().unwrap_or_else(|| txn.check.base.clone());
195        let (s, defs) = self.observe(&manifest);
196        let missed: Vec<String> = txn
197            .expected
198            .units
199            .iter()
200            .filter(|(u, want)| defs.contains_key(*u) && &s.status(u, &defs) != *want)
201            .map(|(u, want)| format!("{u}: expected {want:?}, observed {:?}", s.status(u, &defs)))
202            .collect();
203        if missed.is_empty() { Outcome::Succeeded } else { Outcome::Partial(missed.join("; ")) }
204    }
205
206    fn check(&self, txn: &Transaction, phase: CheckPhase) -> Verdict {
207        let settle = match phase {
208            CheckPhase::Candidate => txn.expected.settle.clone(),
209            CheckPhase::Revert => Settle::Duration(DEFAULT_SETTLE),
210        };
211        self.settle(&settle);
212        let manifest = match phase {
213            CheckPhase::Candidate => txn.check.candidate.clone().unwrap_or_else(|| txn.check.base.clone()),
214            CheckPhase::Revert => txn.check.base.clone(),
215        };
216        // Re-observe for a while: units the activation restarted may still be
217        // coming up. A definite failure that persists past the patience window fails.
218        let deadline = self.b.clock().now().plus(CHECK_PATIENCE);
219        loop {
220            let (s, defs) = self.observe(&manifest);
221            let v = self.violations(txn, phase, &s, &defs);
222            if v.is_empty() {
223                return Verdict::Pass;
224            }
225            if self.b.clock().now() >= deadline {
226                return Verdict::Fail(v.join("; "));
227            }
228            self.b.clock().sleep(CHECK_POLL);
229        }
230    }
231
232    fn release_gc_root(&self, path: &str) -> bool {
233        self.b.generations().release_root(path)
234    }
235
236    fn now(&self) -> Timestamp {
237        self.b.clock().now()
238    }
239
240    fn boot_now(&self) -> Timestamp {
241        self.b.clock().boot_now()
242    }
243
244    fn sleep(&self, d: Millis) {
245        self.b.clock().sleep(d)
246    }
247
248    fn kill_worker(&self, pid: u32) -> bool {
249        self.b.processes().kill_group(pid)
250    }
251
252    fn pid_alive(&self, pid: u32) -> bool {
253        self.b.processes().alive(pid)
254    }
255}
256
257/// `u` fails only free-space probes.
258fn space_only(u: &UnitId, s: &SystemState, defs: &UnitDefs) -> bool {
259    let failing = s.failing_readiness(u, defs);
260    !failing.is_empty()
261        && defs.get(u).is_some_and(|d| {
262            failing.iter().all(|f| d.probes.iter().any(|p| &p.id == f && matches!(p.kind, ProbeKind::FreeSpace { .. })))
263        })
264}
265
266/// A manifest entry as the manifest declares it: the fields the module
267/// derives cleared. Derived (computed by `nix/module.nix` from the unit's own
268/// systemd config, exported, never a `services.planarian.units` option):
269/// `holds`. A change in a derived field alone is no edit to Planarian's
270/// manifest (coordinator, Nov 3); the systemd config behind it is caught
271/// where it already is. Every field named, with no `..`: a field added to
272/// `UnitDef` does not compile here until it is put on one side.
273pub fn declared(d: &UnitDef) -> UnitDef {
274    let UnitDef {
275        id,
276        desired,
277        edges,
278        sensitivity,
279        restart_disruption,
280        probes,
281        synonyms,
282        systemd_restart,
283        startup_grace,
284        runtime_repair,
285        stops_are_faults,
286        hold_allow,
287        hold_interfaces,
288        // Derived:
289        holds: _,
290    } = d.clone();
291    UnitDef {
292        id,
293        desired,
294        edges,
295        sensitivity,
296        restart_disruption,
297        probes,
298        synonyms,
299        systemd_restart,
300        startup_grace,
301        runtime_repair,
302        stops_are_faults,
303        hold_allow,
304        hold_interfaces,
305        holds: false,
306    }
307}
308
309/// Units whose entries, as declared ([`declared`]), differ between two
310/// manifests, or that exist in only one. Only system-scope entries
311/// (`Manifest::units`) are compared; a change to a user-scope template
312/// (`user_units`) is `touches_manifest`'s alone.
313pub fn changed_units(a: &Manifest, b: &Manifest) -> BTreeSet<UnitId> {
314    let mut out = BTreeSet::new();
315    for (id, d) in &a.units {
316        if b.units.get(id).map(declared) != Some(declared(d)) {
317            out.insert(id.clone());
318        }
319    }
320    for id in b.units.keys() {
321        if !a.units.contains_key(id) {
322            out.insert(id.clone());
323        }
324    }
325    out
326}
327
328// Catch-up property tests (docs/regression-ledger.md: C4, C5).
329#[cfg(test)]
330#[path = "host_props.rs"]
331mod props;
planarian_exec/host.rs

planarian_exec/
host.rs
