clawft_kernel/workload_ctl/
plane_lifecycle.rs

1//! Controller-side lifecycle (ADR-099 section 7): mirror each instance's
2//! state from its node, notice a `Dead` node, mark its instances `Lost` and
3//! reschedule them through the same gate and placement engine as any
4//! placement, and settle with a node that comes back.
5//!
6//! **Delivery is at-least-once, with no fencing.** While a node is cut off
7//! but still running, its copy keeps running (and a sensor cog keeps
8//! producing) until the controller notices, replaces it and the node
9//! returns to be told to unload the old one. For the length of the
10//! partition the workload may run twice. The opt-in node lease
11//! (`WorkloadHostService::with_lease`) narrows that window: a node that has
12//! heard no controller for the lease stops its continuous instances itself.
13//! See ADR-099 section 7.
14//!
15//! [`PlacementControlPlane::lifecycle_tick`] is one pass:
16//!
17//! 1. re-describe every target (reachability) and note how long each has
18//!    been down;
19//! 2. a node is `Dead` when it has been unreachable for
20//!    `PlaneConfig::dead_after` and membership, when it knows the peer, does
21//!    not still hold it as healthy. Direct contact is the authority:
22//!    membership never makes a node dead that answers;
23//! 3. each instance on a `Dead` node becomes `Lost` and is rescheduled
24//!    (`plane_reschedule`) unless it is pinned, non-migratable (operator,
25//!    kind, or needing attached hardware), has no stored order, or moved or
26//!    failed to move too often; those raise an alert (chained) and stay
27//!    `Lost`;
28//! 4. the old record of a rescheduled instance is kept as `Rescheduled`.
29//!    When its node returns, that copy is unloaded there (signed, gated,
30//!    chained), with backoff and one chained report per distinct failure;
31//! 5. a node that is reachable but no longer holds an instance is told
32//!    apart: it reports explicitly what a revocation or an unload took
33//!    down, and anything else was lost with the node (a reboot) and is
34//!    placed again;
35//! 6. an instance that was `Lost` and is still on its node when it returns
36//!    is adopted back (a lease-stopped one is restarted).
37//!
38//! Every step is chained: `workload.lifecycle` for state changes and alerts,
39//! `workload.migrate` for a reschedule.
40
41use serde::{Deserialize, Serialize};
42use serde_json::{Value, json};
43
44use super::lifecycle::{InstanceLife, LifecyclePolicy, LifecycleState};
45use super::msg::{RefusalCode, method};
46use super::plane::{CallFailure, PlacementControlPlane, PlacementRecord, PlaneError, now_ms};
47use super::plane_place::PlaceOrder;
48use crate::chain;
49
50/// What the controller keeps per instance beyond its placement record.
51#[derive(Debug, Clone, Serialize, Deserialize)]
52pub struct ControllerLife {
53    /// Current state and history.
54    pub life: InstanceLife,
55    /// May it move, and how often.
56    pub policy: LifecyclePolicy,
57    /// The order it was placed with (what a reschedule re-runs).
58    #[serde(default)]
59    pub order: Option<PlaceOrder>,
60    /// Times it was moved to another node.
61    #[serde(default)]
62    pub reschedules: u32,
63    /// Last reason a reschedule could not happen (chained once per change).
64    #[serde(default)]
65    pub last_error: Option<String>,
66    /// It was meant to be running when its node was lost. A stopped,
67    /// failed or finished instance is never started somewhere else.
68    #[serde(default = "yes")]
69    pub was_running: bool,
70    /// Reschedule attempts that found no node (bounded by `max_reschedules`).
71    #[serde(default)]
72    pub attempts: u32,
73    /// Earliest time of the next reschedule attempt (ms).
74    #[serde(default)]
75    pub next_attempt_ms: u64,
76    /// Failed unloads of the replaced copy on its returned node.
77    #[serde(default)]
78    pub orphan_attempts: u32,
79    /// Earliest time of the next orphan unload (ms).
80    #[serde(default)]
81    pub orphan_next_ms: u64,
82    /// Last orphan-unload failure (chained once per change).
83    #[serde(default)]
84    pub orphan_error: Option<String>,
85}
86
87fn yes() -> bool {
88    true
89}
90
91impl ControllerLife {
92    pub(super) fn new(
93        state: LifecycleState,
94        policy: LifecyclePolicy,
95        order: Option<PlaceOrder>,
96        reschedules: u32,
97    ) -> Self {
98        Self {
99            life: InstanceLife::new(state, now_ms()),
100            policy,
101            order,
102            reschedules,
103            last_error: None,
104            was_running: true,
105            attempts: 0,
106            next_attempt_ms: 0,
107            orphan_attempts: 0,
108            orphan_next_ms: 0,
109            orphan_error: None,
110        }
111    }
112}
113
114/// Something one [`PlacementControlPlane::lifecycle_tick`] did.
115#[derive(Debug, Clone, PartialEq, Serialize)]
116pub struct LifecycleEvent {
117    /// Instance id (on the node it was last placed on).
118    pub instance_id: String,
119    /// That node.
120    pub node_id: String,
121    /// `lost`, `rescheduled`, `replaced`, `alert`, `reschedule_failed`,
122    /// `orphan_unloaded`, `orphan_unload_failed`, `recovered`, `failed` or
123    /// `gone`.
124    pub action: String,
125    /// Human-readable detail.
126    pub detail: String,
127}
128
129pub(super) fn event(rec: &PlacementRecord, action: &str, detail: impl Into<String>) -> LifecycleEvent {
130    LifecycleEvent {
131        instance_id: rec.instance_id.clone(),
132        node_id: rec.node_id.clone(),
133        action: action.into(),
134        detail: detail.into(),
135    }
136}
137
138/// Longest wait between retries of a failed reschedule or orphan unload.
139const MAX_BACKOFF_MS: u64 = 300_000;
140
141pub(super) fn backoff_ms(base_ms: u64, attempts: u32) -> u64 {
142    base_ms
143        .saturating_mul(1u64 << attempts.saturating_sub(1).min(20))
144        .min(MAX_BACKOFF_MS)
145}
146
147impl PlacementControlPlane {
148    /// Remember how a fresh placement was ordered.
149    pub(super) fn register_life(&self, rec: &PlacementRecord, order: &PlaceOrder, p: LifecyclePolicy) {
150        let state = if order.start {
151            LifecycleState::Running
152        } else {
153            LifecycleState::Loaded
154        };
155        if let Ok(mut l) = self.lives.lock() {
156            l.insert(
157                rec.instance_id.clone(),
158                ControllerLife::new(state, p, Some(order.clone()), 0),
159            );
160        }
161        self.persist();
162    }
163
164    pub(super) fn drop_life(&self, instance_id: &str) {
165        if let Ok(mut l) = self.lives.lock() {
166            l.remove(instance_id);
167        }
168    }
169
170    /// Lifecycle state the controller holds for `instance_id`.
171    pub fn lifecycle_of(&self, instance_id: &str) -> Option<LifecycleState> {
172        self.lives
173            .lock()
174            .ok()
175            .and_then(|l| l.get(instance_id).map(|c| c.life.state))
176    }
177
178    /// `{state, restarts, reschedules, has_error}` the controller holds for
179    /// `instance_id`; in memory only, no peer is contacted. The error text is
180    /// withheld (it can carry paths and peer detail).
181    pub fn life_summary(&self, instance_id: &str) -> Option<Value> {
182        self.life_of(instance_id).map(|c| {
183            json!({
184                "state": c.life.state,
185                "restarts": c.life.restarts.len(),
186                "reschedules": c.reschedules,
187                "has_error": c.last_error.is_some(),
188            })
189        })
190    }
191
192    pub(super) fn life_of(&self, instance_id: &str) -> Option<ControllerLife> {
193        self.lives.lock().ok().and_then(|l| l.get(instance_id).cloned())
194    }
195
196    pub(super) fn update_life(&self, instance_id: &str, f: impl FnOnce(&mut ControllerLife)) {
197        if let Ok(mut l) = self.lives.lock()
198            && let Some(c) = l.get_mut(instance_id)
199        {
200            f(c);
201        }
202    }
203
204    pub(super) fn set_state(&self, rec: &PlacementRecord, to: LifecycleState, reason: &str, force: bool) {
205        let kind = chain::EVENT_KIND_WORKLOAD_LIFECYCLE;
206        let from = {
207            let Ok(mut l) = self.lives.lock() else { return };
208            let c = l.entry(rec.instance_id.clone()).or_insert_with(|| {
209                ControllerLife::new(LifecycleState::Running, LifecyclePolicy::default(), None, 0)
210            });
211            if to == LifecycleState::Lost {
212                c.was_running = c.life.state.is_live();
213            }
214            match c.life.transition(to, now_ms()) {
215                Ok(f) => f,
216                // What the node reports wins over what the table allows.
217                Err(_) if force => {
218                    let f = c.life.state;
219                    c.life.state = to;
220                    c.life.since_ms = now_ms();
221                    f
222                }
223                Err(_) => return,
224            }
225        };
226        self.chain_event(
227            kind,
228            json!({ "phase": "lifecycle", "instance_id": rec.instance_id, "node": rec.node_id,
229                    "workload": rec.workload, "from": from, "to": to, "reason": reason }),
230        );
231    }
232
233    /// Whether `node` is `Dead` now: unreachable for `dead_after`, and
234    /// membership (when it knows the peer) does not hold it healthy.
235    fn node_dead(&self, node: &str, now: u64) -> bool {
236        let since = self.down_since.lock().ok().and_then(|d| d.get(node).copied());
237        let down = since.is_some_and(|s| now.saturating_sub(s) >= self.cfg.dead_after.as_millis() as u64);
238        if !down {
239            return false;
240        }
241        match self.membership.as_ref().and_then(|m| m.get_peer(node)) {
242            Some(p) => !matches!(
243                super::facts::liveness_of(&p.state),
244                clawft_types::placement::engine::Liveness::Alive
245            ),
246            None => true,
247        }
248    }
249
250    fn note_reachability(&self, now: u64) {
251        let targets = self.targets();
252        let Ok(mut d) = self.down_since.lock() else { return };
253        for t in targets {
254            if t.reachable {
255                d.remove(&t.node_id);
256            } else {
257                d.entry(t.node_id).or_insert(now);
258            }
259        }
260    }
261
262    /// One pass of the controller lifecycle; see the module docs.
263    pub async fn lifecycle_tick(&self) -> Vec<LifecycleEvent> {
264        self.refresh().await;
265        let now = now_ms();
266        self.note_reachability(now);
267        let mut out = Vec::new();
268        self.mirror_states(&mut out).await;
269        let known: Vec<String> = self.targets().into_iter().map(|t| t.node_id).collect();
270        for rec in self.placements() {
271            // Seed placements and instances on nodes we have no target for
272            // are not this pass's business (a Seed is hardware-bound).
273            if !known.contains(&rec.node_id) {
274                continue;
275            }
276            let Some(state) = self.lifecycle_of(&rec.instance_id) else {
277                self.set_state(&rec, LifecycleState::Running, "adopted by the lifecycle", false);
278                continue;
279            };
280            let dead = self.node_dead(&rec.node_id, now);
281            match (dead, state) {
282                (false, LifecycleState::Rescheduled) => self.unload_orphan(&rec, &mut out).await,
283                (_, LifecycleState::Rescheduled) => {}
284                (true, LifecycleState::Lost) => self.reschedule(&rec, true, &mut out).await,
285                (true, s) if !s.is_terminal() => {
286                    self.set_state(&rec, LifecycleState::Lost, "node is dead", false);
287                    out.push(event(&rec, "lost", format!("node {} is dead", rec.node_id)));
288                    self.reschedule(&rec, true, &mut out).await;
289                }
290                (false, LifecycleState::Lost) => self.rejoined(&rec, &mut out).await,
291                _ => {}
292            }
293        }
294        self.persist();
295        out
296    }
297
298    /// Mirror each reachable node's instance states. An instance the node no
299    /// longer holds is a real teardown when the node says so (revoked,
300    /// unloaded), and otherwise was lost with the node (a reboot): it is
301    /// placed again.
302    async fn mirror_states(&self, out: &mut Vec<LifecycleEvent>) {
303        let recs = self.placements();
304        for t in self.targets().into_iter().filter(|t| t.reachable) {
305            let mine: Vec<&PlacementRecord> = recs.iter().filter(|r| r.node_id == t.node_id).collect();
306            if mine.is_empty() {
307                continue;
308            }
309            let body = json!({ "include_departed": true });
310            let Ok(Value::Array(rows)) = self.call(&t.node_id, method::STATUS, None, body).await else {
311                continue;
312            };
313            for rec in mine {
314                let state = self.lifecycle_of(&rec.instance_id);
315                if matches!(state, Some(LifecycleState::Rescheduled | LifecycleState::Lost)) {
316                    continue; // handled by the rejoin path
317                }
318                let row = rows
319                    .iter()
320                    .find(|r| r["instance_id"].as_str() == Some(rec.instance_id.as_str()));
321                match row {
322                    None => self.missing_on_node(rec, &rows, state, out).await,
323                    Some(r) if r["lease_stopped"] == true => {
324                        // The node stopped it because it heard no controller;
325                        // one is here now. Start it (never mirror this as an
326                        // operator stop).
327                        match self.instance(method::START, &rec.instance_id).await {
328                            Ok(_) => {
329                                self.set_state(rec, LifecycleState::Running, "restarted after a lease stop", true);
330                                out.push(event(rec, "recovered", "restarted after the node's lease stopped it"));
331                            }
332                            Err(e) => out.push(event(rec, "alert", format!("lease-stopped instance not restarted: {e}"))),
333                        }
334                    }
335                    Some(r) => {
336                        let to = serde_json::from_value::<LifecycleState>(r["lifecycle"].clone()).ok();
337                        if let Some(to) = to
338                            && Some(to) != state
339                        {
340                            self.set_state(rec, to, "reported by its node", true);
341                        }
342                    }
343                }
344            }
345        }
346    }
347
348    async fn missing_on_node(
349        &self,
350        rec: &PlacementRecord,
351        rows: &[Value],
352        state: Option<LifecycleState>,
353        out: &mut Vec<LifecycleEvent>,
354    ) {
355        let told = rows
356            .iter()
357            .find(|r| r["departed_instance"].as_str() == Some(rec.instance_id.as_str()))
358            .and_then(|r| r["reason"].as_str());
359        let was_live = state.is_some_and(|s| s.is_live() || s == LifecycleState::Loaded);
360        if told.is_none() && was_live {
361            self.set_state(rec, LifecycleState::Lost, "instance missing though its node is up (restart?)", false);
362            out.push(event(rec, "lost", "the node answers but no longer holds it"));
363            self.reschedule(rec, false, out).await;
364            return;
365        }
366        let why = told.unwrap_or("not running when it went");
367        self.set_state(rec, LifecycleState::Unloaded, &format!("no longer on its node: {why}"), false);
368        self.forget_instances(std::slice::from_ref(&rec.instance_id));
369        out.push(event(rec, "gone", format!("the node no longer holds it ({why})")));
370    }
371
372    /// An old record whose node answers again: unload what is left there,
373    /// backing off after a failure and chaining each distinct failure once.
374    async fn unload_orphan(&self, rec: &PlacementRecord, out: &mut Vec<LifecycleEvent>) {
375        let now = now_ms();
376        let Some(c) = self.life_of(&rec.instance_id) else { return };
377        if now < c.orphan_next_ms {
378            return;
379        }
380        let failure = match self.instance(method::UNLOAD, &rec.instance_id).await {
381            Ok(_) => {
382                out.push(event(rec, "orphan_unloaded", "node returned; the replaced copy was unloaded"));
383                return;
384            }
385            Err(PlaneError::Call(CallFailure::Refused(r))) if r.code == RefusalCode::UnknownInstance => {
386                // The node restarted and no longer holds it.
387                self.drop_life(&rec.instance_id);
388                self.forget_instances(std::slice::from_ref(&rec.instance_id));
389                out.push(event(rec, "orphan_unloaded", "node returned without the replaced copy"));
390                return;
391            }
392            Err(e) => e.to_string(),
393        };
394        let attempts = c.orphan_attempts + 1;
395        let next = now + backoff_ms(self.cfg.retry_backoff.as_millis() as u64, attempts);
396        let fresh = c.orphan_error.as_deref() != Some(failure.as_str());
397        self.update_life(&rec.instance_id, |l| {
398            l.orphan_attempts = attempts;
399            l.orphan_next_ms = next;
400            l.orphan_error = Some(failure.clone());
401        });
402        if fresh {
403            self.chain_event(
404                chain::EVENT_KIND_WORKLOAD_LIFECYCLE,
405                json!({ "phase": "alert", "instance_id": rec.instance_id, "node": rec.node_id,
406                        "workload": rec.workload, "attempt": attempts,
407                        "reason": format!("replaced copy not unloaded on the returned node: {failure}") }),
408            );
409            out.push(event(rec, "orphan_unload_failed", failure));
410        }
411    }
412
413    /// A `Lost` instance whose node is back: adopt it if it is still there
414    /// (restarting it when the node's lease stopped it), otherwise place it
415    /// again.
416    async fn rejoined(&self, rec: &PlacementRecord, out: &mut Vec<LifecycleEvent>) {
417        let body = json!({ "include_departed": true });
418        let Ok(Value::Array(rows)) = self.call(&rec.node_id, method::STATUS, None, body).await else {
419            return;
420        };
421        let row = rows
422            .iter()
423            .find(|r| r["instance_id"].as_str() == Some(rec.instance_id.as_str()));
424        let Some(row) = row else {
425            // Taken down on purpose while we could not see it (an unload
426            // during a partition): not undone.
427            if let Some(why) = rows
428                .iter()
429                .find(|r| r["departed_instance"].as_str() == Some(rec.instance_id.as_str()))
430                .and_then(|r| r["reason"].as_str())
431            {
432                self.set_state(rec, LifecycleState::Unloaded, &format!("taken down while unreachable: {why}"), true);
433                self.forget_instances(std::slice::from_ref(&rec.instance_id));
434                out.push(event(rec, "gone", format!("the node took it down while unreachable ({why})")));
435                return;
436            }
437            return self.reschedule(rec, false, out).await;
438        };
439        if row["lease_stopped"] == true
440            && let Err(e) = self.instance(method::START, &rec.instance_id).await
441        {
442            // Retried next tick; the instance stays Lost until it runs.
443            out.push(event(rec, "alert", format!("lease-stopped instance not restarted: {e}")));
444            return;
445        }
446        self.set_state(rec, LifecycleState::Running, "node rejoined with the instance", false);
447        self.update_life(&rec.instance_id, |l| {
448            l.last_error = None;
449            l.attempts = 0;
450        });
451        out.push(event(rec, "recovered", "node rejoined; the instance is still there"));
452    }
453
454    /// Run [`Self::lifecycle_tick`] every `period` for as long as the runtime lives.
455    pub fn spawn_lifecycle(
456        self: &std::sync::Arc<Self>,
457        period: std::time::Duration,
458    ) -> tokio::task::JoinHandle<()> {
459        let me = std::sync::Arc::clone(self);
460        tokio::spawn(async move {
461            let mut t = tokio::time::interval(period);
462            t.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
463            t.tick().await;
464            loop {
465                t.tick().await;
466                let acted = me.lifecycle_tick().await;
467                if !acted.is_empty() {
468                    tracing::warn!(n = acted.len(), "placement lifecycle");
469                }
470            }
471        })
472    }
473}
clawft_kernel/workload_ctl/plane_lifecycle.rs

clawft_kernel/workload_ctl/
plane_lifecycle.rs
