planarian_incident/
lib.rs

1//! Incident lifecycle (SPEC §13): open, defer to systemd, attempts with
2//! budget and backoff, close on a stabilized desired predicate, regression,
3//! escalation, cooldown. Pure: time comes in as an argument, and every
4//! incident that changes is returned so the daemon can journal it.
5
6use planarian_schema::{
7    DesiredStatus, EscalationReason, Incident, IncidentId, IncidentKind, IncidentState, Millis, Outcome, SystemState,
8    Timestamp, TxnId, UnitDefs, UnitId, UnitStatus,
9};
10use serde::{Deserialize, Serialize};
11use std::collections::BTreeMap;
12
13pub const BUDGET: u8 = 3;
14pub const BACKOFF: [Millis; 3] = [Millis::secs(10), Millis::secs(60), Millis::secs(300)];
15pub const STABILIZATION: Millis = Millis::secs(5 * 60);
16pub const REGRESSION_WINDOW: Millis = Millis::secs(30 * 60);
17pub const ESCALATE_AT: u8 = 3;
18pub const COOLDOWN_FIRST: Millis = Millis::secs(3600);
19pub const COOLDOWN_LATER: Millis = Millis::secs(6 * 3600);
20
21/// How long an escalation holds the incident before Planarian looks again
22/// (§13: no autonomous action until the hold ends or someone acts).
23/// - `Regressed`: 1 h, §13's hold for the third regression.
24/// - `RevertBlockedByFloor`: 6 h. §12e names no cooldown; a revert below the
25///   floor needs an administrator, which an hour does not change
26///   (docs/deviations.md).
27/// - The others have no escalation today (budget and diagnoser exhaustion
28///   cool down, `Cooldown`, through `cool`'s 1 h then 6 h; RecoveryFailed sets
29///   the interlock): 1 h, §13's first cooldown, should one arise.
30pub fn escalation_cooldown(reason: EscalationReason) -> Millis {
31    match reason {
32        EscalationReason::Regressed => COOLDOWN_FIRST,
33        EscalationReason::RevertBlockedByFloor => COOLDOWN_LATER,
34        EscalationReason::BudgetExhausted | EscalationReason::DiagnoserExhausted | EscalationReason::RecoveryFailed => {
35            COOLDOWN_FIRST
36        }
37    }
38}
39pub const SETTLE: Millis = Millis::secs(15);
40/// Consecutive ticks of divergence before an incident opens.
41pub const OPEN_AFTER: u8 = 2;
42/// Consecutive Unknown ticks before a diagnostic incident opens (§10).
43pub const UNKNOWN_LIMIT: u8 = 3;
44/// How far back failures count toward systemd's own start limit.
45pub const FAILURE_WINDOW: Millis = Millis::secs(10 * 60);
46
47/// §13 "Open when …": the divergences that open an incident.
48pub fn divergent(desired: DesiredStatus, status: &UnitStatus) -> bool {
49    use UnitStatus as S;
50    match desired {
51        DesiredStatus::Running => {
52            matches!(status, S::Failed(_) | S::Degraded(_) | S::Stopped | S::Absent | S::Detected)
53        }
54        DesiredStatus::Bound => matches!(status, S::Detected | S::Absent | S::Failed(_)),
55        DesiredStatus::Stopped => matches!(status, S::Failed(_)),
56        DesiredStatus::Absent => false,
57    }
58}
59
60/// The desired predicate (§13 "Closed when").
61pub fn satisfied(desired: DesiredStatus, status: &UnitStatus) -> bool {
62    use UnitStatus as S;
63    match desired {
64        DesiredStatus::Running => matches!(status, S::Running),
65        DesiredStatus::Bound => matches!(status, S::Bound | S::Running),
66        DesiredStatus::Stopped => matches!(status, S::Stopped | S::Absent),
67        DesiredStatus::Absent => matches!(status, S::Absent),
68    }
69}
70
71/// Per-unit observation history. Not journaled: after a restart it is
72/// re-learned within a few ticks, while the incidents themselves (budgets,
73/// backoff, holds) come back from the journal.
74#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
75struct Track {
76    divergent_ticks: u8,
77    unknown_ticks: u8,
78    satisfied_since: Option<Timestamp>,
79    /// When the last incident on this target closed, and its regression count.
80    last_closed: Option<(Timestamp, u8)>,
81    /// When the unit went from fine to divergent, recently.
82    failures: Vec<Timestamp>,
83    was_divergent: bool,
84    /// When the unit last came up (Running or Degraded: its process is active),
85    /// or when Planarian started or restarted it; first sight counts as coming up.
86    up_since: Option<Timestamp>,
87    /// Degraded within its startup grace: not yet a fault.
88    in_grace: bool,
89}
90
91#[derive(Clone, Debug)]
92pub struct Manager {
93    pub incidents: BTreeMap<IncidentId, Incident>,
94    tracks: BTreeMap<UnitId, Track>,
95    next_id: u64,
96    /// How long after coming up a failing health probe is not yet a fault,
97    /// for units that do not set their own (`UnitDef::startup_grace`).
98    pub startup_grace: Millis,
99}
100
101/// The default startup grace (`services.planarian.startupGrace`).
102pub const STARTUP_GRACE: Millis = Millis::secs(90);
103
104impl Default for Manager {
105    fn default() -> Manager {
106        Manager::restore([])
107    }
108}
109
110impl Manager {
111    /// Rebuild from the incidents the journal holds (open and cooling ones).
112    pub fn restore(incidents: impl IntoIterator<Item = Incident>) -> Manager {
113        let incidents: BTreeMap<IncidentId, Incident> = incidents.into_iter().map(|i| (i.id, i)).collect();
114        let next_id = incidents.keys().map(|i| i.0).max().unwrap_or(0);
115        Manager { incidents, tracks: BTreeMap::new(), next_id, startup_grace: STARTUP_GRACE }
116    }
117
118    /// Planarian (or a user through it) just started or restarted `unit`:
119    /// its startup grace begins again.
120    pub fn started(&mut self, unit: &UnitId, now: Timestamp) {
121        self.tracks.entry(unit.clone()).or_default().up_since = Some(now);
122    }
123
124    pub fn active_for(&self, unit: &UnitId) -> Option<&Incident> {
125        self.incidents.values().find(|i| &i.target == unit && i.state != IncidentState::Closed)
126    }
127
128    fn active_id(&self, unit: &UnitId) -> Option<IncidentId> {
129        self.active_for(unit).map(|i| i.id)
130    }
131
132    fn close(&mut self, id: IncidentId, now: Timestamp, changed: &mut Vec<Incident>) {
133        if let Some(mut i) = self.incidents.remove(&id) {
134            if i.kind == IncidentKind::Repair {
135                self.tracks.entry(i.target.clone()).or_default().last_closed = Some((now, i.regressions));
136            }
137            i.state = IncidentState::Closed;
138            changed.push(i);
139        }
140    }
141
142    /// Observe one tick. Returns every incident that opened, changed or closed.
143    pub fn tick(&mut self, now: Timestamp, state: &SystemState, defs: &UnitDefs) -> Vec<Incident> {
144        self.tick_at(now, now, state, defs)
145    }
146
147    /// `tick`, with the decision's own time: `now` is the observation's
148    /// (what the evidence is timed by), `decided` the clock read after it,
149    /// once the probes have run. A hold set here (a third regression's
150    /// escalation) runs from `decided`, as every cooldown and backoff does
151    /// (§13), so a slow observation never shortens it.
152    pub fn tick_at(
153        &mut self,
154        now: Timestamp,
155        decided: Timestamp,
156        state: &SystemState,
157        defs: &UnitDefs,
158    ) -> Vec<Incident> {
159        let mut changed = vec![];
160
161        // Targets that left the definitions close.
162        let gone: Vec<IncidentId> =
163            self.incidents.values().filter(|i| !defs.contains_key(&i.target)).map(|i| i.id).collect();
164        for id in gone {
165            self.close(id, now, &mut changed);
166        }
167
168        // Holds that ended.
169        for i in self.incidents.values_mut() {
170            match i.state {
171                IncidentState::Cooldown { until } | IncidentState::Escalated { until, .. } if now >= until => {
172                    i.state = IncidentState::Open;
173                    i.budget_left = 1;
174                    i.next_attempt = now;
175                    changed.push(i.clone());
176                }
177                _ => {}
178            }
179        }
180
181        for (id, def) in defs {
182            let status = state.status(id, defs);
183            let desired = def.desired.status;
184            let active = self.active_id(id);
185
186            if matches!(status, UnitStatus::Overridden(_) | UnitStatus::NotExpected) {
187                // A suppression closes the incident (I13); so does a condition
188                // that stopped holding (logout, unplug).
189                if let Some(a) = active {
190                    self.close(a, now, &mut changed);
191                }
192                let t = self.tracks.entry(id.clone()).or_default();
193                (t.divergent_ticks, t.unknown_ticks, t.satisfied_since, t.was_divergent) = (0, 0, None, false);
194                continue;
195            }
196
197            if status == UnitStatus::Unknown {
198                let t = self.tracks.entry(id.clone()).or_default();
199                t.unknown_ticks = t.unknown_ticks.saturating_add(1);
200                t.divergent_ticks = 0;
201                t.satisfied_since = None;
202                if t.unknown_ticks >= UNKNOWN_LIMIT && active.is_none() {
203                    let inc = self.open(id, IncidentKind::Diagnostic, (desired, status), now, now, 0);
204                    changed.push(inc);
205                }
206                continue;
207            }
208
209            // Startup grace: a unit whose process is up but whose health probe
210            // fails may still be starting. A crash (Failed) gets no grace.
211            {
212                let t = self.tracks.entry(id.clone()).or_default();
213                if matches!(status, UnitStatus::Running | UnitStatus::Degraded(_)) {
214                    t.up_since.get_or_insert(now);
215                } else {
216                    t.up_since = None;
217                }
218                let grace = def.startup_grace.unwrap_or(self.startup_grace);
219                t.in_grace = matches!(status, UnitStatus::Degraded(_))
220                    && t.up_since.is_some_and(|since| now.0.saturating_sub(since.0) < grace.0);
221                if t.in_grace {
222                    (t.divergent_ticks, t.unknown_ticks, t.satisfied_since) = (0, 0, None);
223                    continue;
224                }
225            }
226
227            if divergent(desired, &status) {
228                let t = self.tracks.entry(id.clone()).or_default();
229                t.unknown_ticks = 0;
230                t.satisfied_since = None;
231                t.divergent_ticks = t.divergent_ticks.saturating_add(1);
232                if !t.was_divergent {
233                    t.was_divergent = true;
234                    t.failures.push(now);
235                }
236                t.failures.retain(|f| now.0.saturating_sub(f.0) <= FAILURE_WINDOW.0);
237                let ready = t.divergent_ticks >= OPEN_AFTER;
238                let recent_failures = t.failures.len() as u32;
239                let last_closed = t.last_closed;
240
241                if let Some(a) = active {
242                    let inc = self.incidents.get_mut(&a).expect("active");
243                    if inc.kind == IncidentKind::Diagnostic {
244                        // The evidence turned definite: this is a repair incident now.
245                        inc.kind = IncidentKind::Repair;
246                        inc.budget_left = BUDGET;
247                        inc.next_attempt = now;
248                    }
249                    if inc.divergence.1 != status {
250                        inc.divergence = (desired, status.clone());
251                        changed.push(inc.clone());
252                    }
253                    continue;
254                }
255                if !ready {
256                    continue;
257                }
258                // Defer to systemd while it is within its own start limit (§13).
259                let next_attempt = match &def.systemd_restart {
260                    Some(p) if p.restart && recent_failures <= p.start_limit_burst => {
261                        now.plus(Millis(p.restart_sec.0 + SETTLE.0))
262                    }
263                    _ => now,
264                };
265                let regressions = match last_closed {
266                    Some((closed, n)) if now.0.saturating_sub(closed.0) <= REGRESSION_WINDOW.0 => n + 1,
267                    _ => 0,
268                };
269                let mut inc = self.open(id, IncidentKind::Repair, (desired, status), now, next_attempt, regressions);
270                if regressions >= ESCALATE_AT {
271                    let reason = EscalationReason::Regressed;
272                    inc.state = IncidentState::Escalated { until: decided.plus(escalation_cooldown(reason)), reason };
273                } else if regressions > 0 {
274                    inc.state = IncidentState::Regressed { count: regressions };
275                }
276                self.incidents.insert(inc.id, inc.clone());
277                changed.push(inc);
278                continue;
279            }
280
281            let t = self.tracks.entry(id.clone()).or_default();
282            t.divergent_ticks = 0;
283            t.unknown_ticks = 0;
284            t.was_divergent = false;
285            if satisfied(desired, &status) {
286                let since = *t.satisfied_since.get_or_insert(now);
287                if let Some(a) = active {
288                    let kind = self.incidents[&a].kind;
289                    if kind == IncidentKind::Diagnostic || now.0.saturating_sub(since.0) >= STABILIZATION.0 {
290                        self.close(a, now, &mut changed);
291                    }
292                }
293            } else {
294                t.satisfied_since = None;
295            }
296        }
297        changed
298    }
299
300    fn open(
301        &mut self,
302        target: &UnitId,
303        kind: IncidentKind,
304        divergence: (DesiredStatus, UnitStatus),
305        now: Timestamp,
306        next_attempt: Timestamp,
307        regressions: u8,
308    ) -> Incident {
309        self.next_id += 1;
310        let inc = Incident {
311            id: IncidentId(self.next_id),
312            kind,
313            target: target.clone(),
314            divergence,
315            opened: now,
316            attempts: vec![],
317            budget_left: if kind == IncidentKind::Repair { BUDGET } else { 0 },
318            state: IncidentState::Open,
319            backoff: BACKOFF[0],
320            next_attempt,
321            regressions,
322        };
323        self.incidents.insert(inc.id, inc.clone());
324        inc
325    }
326
327    /// Repair incidents the diagnoser may act on now. An incident whose target
328    /// currently meets its desired state is stabilizing, not actionable: acting
329    /// on it would find nothing to do and wrongly exhaust it.
330    pub fn actionable(&self, now: Timestamp) -> Vec<&Incident> {
331        self.incidents
332            .values()
333            .filter(|i| {
334                i.kind == IncidentKind::Repair
335                    && matches!(i.state, IncidentState::Open | IncidentState::Regressed { .. })
336                    && i.budget_left > 0
337                    && now >= i.next_attempt
338                    && self.tracks.get(&i.target).is_none_or(|t| t.satisfied_since.is_none() && !t.in_grace)
339            })
340            .collect()
341    }
342
343    /// Diagnostic incidents (re-observe through another Readiness probe).
344    pub fn diagnostic(&self) -> Vec<&Incident> {
345        self.incidents.values().filter(|i| i.kind == IncidentKind::Diagnostic).collect()
346    }
347
348    /// Record an attempt's transaction. Spends budget, sets the next backoff
349    /// (longer after regressions), and enters cooldown when the budget is gone.
350    pub fn record_attempt(&mut self, id: IncidentId, txn: TxnId, outcome: Outcome, now: Timestamp) -> Option<Incident> {
351        let i = self.incidents.get_mut(&id)?;
352        i.attempts.push((txn, outcome));
353        i.budget_left = i.budget_left.saturating_sub(1);
354        let step = (BUDGET - i.budget_left - 1) as usize + i.regressions as usize;
355        i.backoff = BACKOFF[step.min(BACKOFF.len() - 1)];
356        i.next_attempt = now.plus(i.backoff);
357        if i.budget_left == 0 {
358            Self::cool(i, now);
359        }
360        Some(i.clone())
361    }
362
363    fn cool(i: &mut Incident, now: Timestamp) {
364        // The first cooldown is 1 h; once the budget has been spent before
365        // (the post-cooldown attempt failed too, or the diagnoser exhausted
366        // again), 6 h.
367        let cooled_before = i.attempts.len() > BUDGET as usize || i.backoff.0 >= COOLDOWN_FIRST.0;
368        let d = if cooled_before { COOLDOWN_LATER } else { COOLDOWN_FIRST };
369        i.backoff = d;
370        i.state = IncidentState::Cooldown { until: now.plus(d) };
371    }
372
373    /// The diagnoser found nothing (A: report to the user). The incident goes
374    /// quiet with a cooldown rather than retrying forever (§1 step 4).
375    pub fn exhausted(&mut self, id: IncidentId, now: Timestamp) -> Option<Incident> {
376        let i = self.incidents.get_mut(&id)?;
377        Self::cool(i, now);
378        Some(i.clone())
379    }
380
381    /// Hand the incident to a person, with why (§13): no autonomous action
382    /// until the hold ends or someone acts on the target.
383    pub fn escalate(&mut self, id: IncidentId, reason: EscalationReason, now: Timestamp) -> Option<Incident> {
384        let i = self.incidents.get_mut(&id)?;
385        i.state = IncidentState::Escalated { until: now.plus(escalation_cooldown(reason)), reason };
386        Some(i.clone())
387    }
388
389    /// A human acted on the target (release, explicit request): lift any hold.
390    pub fn human_acted(&mut self, target: &UnitId, now: Timestamp) -> Option<Incident> {
391        let i = self.incidents.values_mut().find(|i| &i.target == target)?;
392        i.state = IncidentState::Open;
393        i.budget_left = i.budget_left.max(1);
394        i.next_attempt = now;
395        Some(i.clone())
396    }
397}
398
399#[cfg(test)]
400mod tests;
planarian_incident/lib.rs

planarian_incident/
lib.rs
