clawft_kernel/workload_ctl/
lifecycle.rs

1//! The instance state machine and the bounded-restart policy (ADR-099
2//! section 7). Pure: no I/O, no clock of its own (callers pass `now_ms`), so
3//! every rule is table-testable.
4//!
5//! The same states serve both sides. A node's `workload-host` walks an
6//! instance through `Loaded`, `Running`, `Unhealthy`, `Restarting`, `Failed`
7//! and the teardown states; the controller adds `Lost` (its node is
8//! `Dead`) and `Rescheduled` (a replacement was placed elsewhere).
9
10use serde::{Deserialize, Serialize};
11
12/// Where an instance is in its life.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
14#[serde(rename_all = "snake_case")]
15pub enum LifecycleState {
16    /// The controller asked for it.
17    Requested,
18    /// A node accepted the placement.
19    Placed,
20    /// The node is fetching the payload.
21    Fetching,
22    /// The payload verified.
23    Verified,
24    /// Loaded, not running.
25    Loaded,
26    /// Running and healthy.
27    Running,
28    /// Missed its health heartbeat; a restart is pending.
29    Unhealthy,
30    /// A restart is under way.
31    Restarting,
32    /// A stop was requested.
33    Stopping,
34    /// Stopped by the operator (not restarted).
35    Stopped,
36    /// A one-shot run ended cleanly.
37    Finished,
38    /// Restart budget spent, or it could not be started. Held for the
39    /// operator; not retried by itself.
40    Failed,
41    /// Its node is `Dead`; the placer decides what happens next.
42    Lost,
43    /// A replacement was placed on another node. The old record only
44    /// waits for its node to return so the orphan can be unloaded.
45    Rescheduled,
46    /// A revocation named it; it is being (or was) taken down.
47    Revoked,
48    /// Gone.
49    Unloaded,
50}
51
52impl LifecycleState {
53    /// Wire name.
54    pub fn as_str(self) -> &'static str {
55        match self {
56            Self::Requested => "requested",
57            Self::Placed => "placed",
58            Self::Fetching => "fetching",
59            Self::Verified => "verified",
60            Self::Loaded => "loaded",
61            Self::Running => "running",
62            Self::Unhealthy => "unhealthy",
63            Self::Restarting => "restarting",
64            Self::Stopping => "stopping",
65            Self::Stopped => "stopped",
66            Self::Finished => "finished",
67            Self::Failed => "failed",
68            Self::Lost => "lost",
69            Self::Rescheduled => "rescheduled",
70            Self::Revoked => "revoked",
71            Self::Unloaded => "unloaded",
72        }
73    }
74
75    /// Nothing follows `Unloaded`.
76    pub fn is_terminal(self) -> bool {
77        self == Self::Unloaded
78    }
79
80    /// The instance is meant to be doing work.
81    pub fn is_live(self) -> bool {
82        matches!(self, Self::Running | Self::Unhealthy | Self::Restarting)
83    }
84}
85
86/// Whether `from -> to` is a legal step.
87pub fn can_transition(from: LifecycleState, to: LifecycleState) -> bool {
88    use LifecycleState::*;
89    if from == to || from.is_terminal() {
90        return false;
91    }
92    // A revocation and a node loss can interrupt anything that exists; a
93    // replaced record (`Rescheduled`) is only ever unloaded.
94    if from != Rescheduled && matches!(to, Revoked | Lost) {
95        return !(from == Revoked && to == Lost);
96    }
97    matches!(
98        (from, to),
99        (Requested, Placed | Failed)
100            | (Placed, Fetching | Loaded | Running | Failed)
101            | (Fetching, Verified | Failed)
102            | (Verified, Loaded | Failed)
103            | (Loaded, Running | Unloaded | Failed)
104            | (Running, Unhealthy | Stopping | Stopped | Finished | Failed | Unloaded)
105            | (Unhealthy, Running | Restarting | Stopping | Failed | Unloaded)
106            | (Restarting, Running | Unhealthy | Failed | Stopping)
107            | (Stopping, Stopped | Unloaded | Failed)
108            | (Stopped, Running | Unloaded | Failed)
109            | (Finished, Running | Unloaded)
110            | (Failed, Running | Unloaded)
111            | (Lost, Running | Rescheduled | Unloaded | Failed)
112            | (Rescheduled, Unloaded)
113            | (Revoked, Unloaded | Failed)
114    )
115}
116
117/// A step that is not allowed.
118#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
119#[error("illegal lifecycle step {from:?} -> {to:?}")]
120pub struct IllegalTransition {
121    /// Where it was.
122    pub from: LifecycleState,
123    /// Where it was asked to go.
124    pub to: LifecycleState,
125}
126
127/// Bounded restarts: at most `max_restarts` inside `window_ms`, spaced by an
128/// exponential backoff.
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
130pub struct RestartPolicy {
131    /// Restarts allowed inside the window (0 disables restarting).
132    pub max_restarts: u32,
133    /// The window restarts are counted in.
134    pub window_ms: u64,
135    /// Delay before the first restart; doubles with each one in the window.
136    pub backoff_base_ms: u64,
137    /// The backoff never exceeds this.
138    pub backoff_max_ms: u64,
139}
140
141impl Default for RestartPolicy {
142    fn default() -> Self {
143        Self {
144            max_restarts: 3,
145            window_ms: 600_000,
146            backoff_base_ms: 1_000,
147            backoff_max_ms: 30_000,
148        }
149    }
150}
151
152/// What to do about an unhealthy instance right now.
153#[derive(Debug, Clone, Copy, PartialEq, Eq)]
154pub enum RestartDecision {
155    /// Restart it now; this is attempt `attempt` (1-based) in the window.
156    Restart {
157        /// Attempt number inside the window.
158        attempt: u32,
159    },
160    /// Not yet: the backoff ends at `until_ms`.
161    Wait {
162        /// Earliest time to restart.
163        until_ms: u64,
164    },
165    /// The budget is spent: mark it failed.
166    Exhausted,
167}
168
169/// One instance's life on one side of the wire.
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct InstanceLife {
172    /// Current state.
173    pub state: LifecycleState,
174    /// When it entered the state (ms).
175    pub since_ms: u64,
176    /// Consecutive bad polls.
177    pub misses: u32,
178    /// Times of the restarts still inside the window (ms, ascending).
179    pub restarts: Vec<u64>,
180    /// When the last poll happened (ms); `0` before the first.
181    pub last_poll_ms: u64,
182}
183
184impl InstanceLife {
185    /// A new life in `state`.
186    pub fn new(state: LifecycleState, now_ms: u64) -> Self {
187        Self {
188            state,
189            since_ms: now_ms,
190            misses: 0,
191            restarts: Vec::new(),
192            last_poll_ms: 0,
193        }
194    }
195
196    /// Take a legal step (the miss count resets on entering `Running`).
197    pub fn transition(
198        &mut self,
199        to: LifecycleState,
200        now_ms: u64,
201    ) -> Result<LifecycleState, IllegalTransition> {
202        let from = self.state;
203        if !can_transition(from, to) {
204            return Err(IllegalTransition { from, to });
205        }
206        self.state = to;
207        self.since_ms = now_ms;
208        if to == LifecycleState::Running {
209            self.misses = 0;
210        }
211        Ok(from)
212    }
213
214    /// Whether `policy` allows a restart at `now_ms`. Restarts older than
215    /// the window no longer count. Does not record anything.
216    pub fn restart_decision(&self, policy: &RestartPolicy, now_ms: u64) -> RestartDecision {
217        let recent: Vec<u64> = self
218            .restarts
219            .iter()
220            .copied()
221            .filter(|t| now_ms.saturating_sub(*t) < policy.window_ms)
222            .collect();
223        let n = recent.len() as u32;
224        if n >= policy.max_restarts {
225            return RestartDecision::Exhausted;
226        }
227        // The wait before restart n+1 is the base for n = 1, then doubles.
228        let shift = n.saturating_sub(1).min(20);
229        let backoff = policy
230            .backoff_base_ms
231            .saturating_mul(1u64 << shift)
232            .min(policy.backoff_max_ms);
233        match recent.last() {
234            Some(last) if now_ms < last.saturating_add(backoff) => RestartDecision::Wait {
235                until_ms: last.saturating_add(backoff),
236            },
237            _ => RestartDecision::Restart { attempt: n + 1 },
238        }
239    }
240
241    /// Record a restart at `now_ms` (and forget those outside the window).
242    pub fn record_restart(&mut self, policy: &RestartPolicy, now_ms: u64) {
243        self.restarts
244            .retain(|t| now_ms.saturating_sub(*t) < policy.window_ms);
245        self.restarts.push(now_ms);
246    }
247}
248
249/// Operator-visible lifecycle settings of one placement.
250#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
251pub struct LifecyclePolicy {
252    /// May the placer move it when its node is lost? `false` raises an
253    /// alert instead. A pin (`PlaceOrder::pin`) also holds it in place.
254    pub migratable: bool,
255    /// Bound on how many times one instance is rescheduled (a flapping
256    /// mesh must not move a workload forever).
257    pub max_reschedules: u32,
258}
259
260impl Default for LifecyclePolicy {
261    fn default() -> Self {
262        Self {
263            migratable: true,
264            max_reschedules: 5,
265        }
266    }
267}
268
269/// Capability namespaces that name something physically attached to or
270/// local to one node: a sensor, a device, a data feed on a node's LAN.
271/// `accel` is deliberately not here: any node with the same accelerator is
272/// as good a home, so failover is the point. `trust`, `store`, `model`,
273/// `cpu`, `os`, `runtime`, `mem`, `node`, `perf` are not hardware either.
274const HARDWARE_NAMESPACES: &[&str] = &["sensor", "device", "feed"];
275
276/// Whether `spec` needs something attached to (or local to) a particular
277/// node: any requirement in [`HARDWARE_NAMESPACES`]. Such a workload's data
278/// source is tied to its node, so moving it is not a failover; it is held
279/// and alerted instead. An allow-list: a new namespace never pins by
280/// accident.
281pub fn needs_attached_hardware(spec: &clawft_types::placement::engine::WorkloadSpec) -> bool {
282    use clawft_types::placement::IdSelector;
283    let reqs = spec
284        .requirements
285        .common
286        .iter()
287        .chain(spec.requirements.variants.iter().flat_map(|v| v.requirements.iter()));
288    reqs.into_iter().any(|r| {
289        let id = match &r.selector {
290            IdSelector::Exact(c) => c.as_str(),
291            IdSelector::Prefix(p) => p.as_str(),
292        };
293        let ns = id.split('.').next().unwrap_or(id);
294        HARDWARE_NAMESPACES.contains(&ns)
295    })
296}
clawft_kernel/workload_ctl/lifecycle.rs

clawft_kernel/workload_ctl/
lifecycle.rs
