harmos_runtime/supervision/
instance.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::sync::{Arc, Mutex, Once, Weak};
18
19// Third Party Imports
20use tokio::sync::watch;
21use tokio::time::Instant;
22
23// Local Imports
24use super::cleanup::{Cleanup, Finalize, Resource};
25use super::executor;
26use crate::Error;
27
28/* === Constants === */
29
30/* === Macros === */
31
32/* === Type Aliases === */
33
34/* === Structs === */
35
36#[derive(Clone, Debug)]
37pub struct Status {
38    phase: Phase,
39    outcome: Option<Outcome>,
40    cancellation_requested: bool,
41    dependents: usize,
42    cleanup_pending: bool,
43    cleanup_failure: Option<Arc<Error>>,
44    attempt: usize,
45    started_at: Option<Instant>,
46    settled_at: Option<Instant>,
47}
48
49pub(crate) struct Instance {
50    status: watch::Sender<Status>,
51    // Resident instances end Stopped; request-scoped ones end Settled.
52    terminal: Phase,
53    finished: Once,
54    on_settled: Option<Box<dyn Fn() + Send + Sync>>,
55    cancellation: watch::Sender<bool>,
56    children: Mutex<Vec<Weak<Instance>>>,
57    retained: Mutex<Option<Arc<dyn Settlement>>>,
58    providers: Mutex<Vec<Dependency>>,
59}
60
61struct Dependency {
62    provider: Arc<Instance>,
63}
64
65/* === Enums === */
66
67#[derive(Clone, Copy, Debug, PartialEq, Eq)]
68pub enum Phase {
69    Queued,
70    Initializing,
71    Running,
72    RetryWaiting,
73    Restarting,
74    Stopping,
75    Finalizing,
76    Stopped,
77    Settled,
78}
79
80#[derive(Clone, Copy, Debug, PartialEq, Eq)]
81pub enum Outcome {
82    Succeeded,
83    Failed,
84    Cancelled,
85    TimedOut,
86    Unavailable,
87}
88
89/* === Traits === */
90
91trait Settlement: Send + Sync {
92    fn settle(&self) -> Finalize<'_>;
93}
94
95/* === Implementations === */
96
97impl Status {
98    pub fn phase(&self) -> Phase {
99        self.phase
100    }
101
102    pub fn outcome(&self) -> Option<Outcome> {
103        self.outcome
104    }
105
106    pub fn cancellation_requested(&self) -> bool {
107        self.cancellation_requested
108    }
109
110    pub fn dependents(&self) -> usize {
111        self.dependents
112    }
113
114    pub fn cleanup_pending(&self) -> bool {
115        self.cleanup_pending
116    }
117
118    pub fn cleanup_failure(&self) -> Option<&Arc<Error>> {
119        self.cleanup_failure.as_ref()
120    }
121
122    pub fn attempt(&self) -> usize {
123        self.attempt
124    }
125
126    pub fn started_at(&self) -> Option<Instant> {
127        self.started_at
128    }
129
130    pub fn settled_at(&self) -> Option<Instant> {
131        self.settled_at
132    }
133
134    // Project onto the status every host binding reports.
135    pub fn portable(&self) -> harmos_domain::operations::instance::Status {
136        use harmos_domain::operations::instance;
137        let phase = match self.phase {
138            Phase::Queued => instance::Phase::Queued,
139            Phase::Initializing => instance::Phase::Initializing,
140            Phase::Running => instance::Phase::Running,
141            Phase::RetryWaiting => instance::Phase::RetryWaiting,
142            Phase::Restarting => instance::Phase::Restarting,
143            Phase::Stopping => instance::Phase::Stopping,
144            Phase::Finalizing => instance::Phase::CleaningUp,
145            Phase::Stopped => instance::Phase::Stopped,
146            Phase::Settled => instance::Phase::Settled,
147        };
148        let outcome = self.outcome.map(|outcome| match outcome {
149            Outcome::Succeeded => instance::Outcome::Succeeded,
150            Outcome::Failed => instance::Outcome::Failed,
151            Outcome::Cancelled => instance::Outcome::Cancelled,
152            Outcome::TimedOut => instance::Outcome::TimedOut,
153            Outcome::Unavailable => instance::Outcome::Unavailable,
154        });
155        instance::Status {
156            phase,
157            outcome,
158            cancellation_requested: self.cancellation_requested,
159            attempt: u32::try_from(self.attempt).unwrap_or(u32::MAX),
160        }
161    }
162}
163
164impl Phase {
165    pub fn is_terminal(self) -> bool {
166        matches!(self, Self::Stopped | Self::Settled)
167    }
168}
169
170impl Instance {
171    pub(crate) fn queued(cancellation: watch::Sender<bool>) -> Self {
172        let status = Status {
173            phase: Phase::Queued,
174            outcome: None,
175            cancellation_requested: false,
176            dependents: 0,
177            cleanup_pending: false,
178            cleanup_failure: None,
179            attempt: 0,
180            started_at: None,
181            settled_at: None,
182        };
183        Self {
184            status: watch::channel(status).0,
185            terminal: Phase::Settled,
186            finished: Once::new(),
187            on_settled: None,
188            cancellation,
189            children: Mutex::new(Vec::new()),
190            retained: Mutex::new(None),
191            providers: Mutex::new(Vec::new()),
192        }
193    }
194
195    // A resident instance runs until stopped and ends in `Stopped`.
196    pub(crate) fn resident(mut self) -> Self {
197        self.terminal = Phase::Stopped;
198        self
199    }
200
201    pub(crate) fn status(&self) -> Status {
202        self.status.borrow().clone()
203    }
204
205    pub(crate) fn settlement(&self) -> watch::Receiver<Status> {
206        self.status.subscribe()
207    }
208
209    pub(crate) fn transition(&self, phase: Phase) {
210        assert!(
211            !matches!(phase, Phase::Queued | Phase::Stopped | Phase::Settled),
212            "admission and physical settlement own terminal transitions"
213        );
214        let now = executor::now();
215        self.status.send_modify(|status| {
216            if status.phase.is_terminal() {
217                return;
218            }
219            if status.cancellation_requested
220                && matches!(phase, Phase::Initializing | Phase::Running | Phase::RetryWaiting | Phase::Restarting)
221            {
222                return;
223            }
224            status.phase = phase;
225            if phase == Phase::Initializing {
226                status.started_at.get_or_insert(now);
227            }
228            status.cleanup_pending = matches!(phase, Phase::Stopping | Phase::Finalizing);
229        });
230    }
231
232    pub(crate) fn attempt(&self, attempt: usize) -> bool {
233        self.begin_attempt(attempt, Phase::Running)
234    }
235
236    pub(crate) fn initialize_attempt(&self, attempt: usize) -> bool {
237        self.begin_attempt(attempt, Phase::Initializing)
238    }
239
240    fn begin_attempt(&self, attempt: usize, phase: Phase) -> bool {
241        let now = executor::now();
242        let mut admitted = false;
243        self.status.send_modify(|status| {
244            if status.cancellation_requested || status.phase.is_terminal() {
245                return;
246            }
247            status.phase = phase;
248            status.cleanup_pending = false;
249            status.attempt = attempt;
250            status.started_at.get_or_insert(now);
251            admitted = true;
252        });
253        admitted
254    }
255
256    pub(crate) fn complete(&self, outcome: Outcome) {
257        self.status.send_modify(|status| {
258            if !status.phase.is_terminal() {
259                status.outcome = Some(outcome);
260            }
261        });
262    }
263
264    pub(crate) fn on_settled(mut self, notify: impl Fn() + Send + Sync + 'static) -> Self {
265        self.on_settled = Some(Box::new(notify));
266        self
267    }
268
269    pub(crate) fn finished(&self) {
270        self.finished.call_once(|| {
271            let providers = std::mem::take(&mut *self.providers.lock().expect("instance providers are healthy"));
272            drop(providers);
273            if let Some(notify) = &self.on_settled {
274                notify();
275            }
276            let now = executor::now();
277            let mut failure = None;
278            self.status.send_modify(|status| {
279                status.phase = self.terminal;
280                status.cleanup_pending = false;
281                failure = status.cleanup_failure.take();
282                status.settled_at = Some(now);
283            });
284            drop(failure);
285        });
286    }
287
288    pub(crate) fn interrupted(&self, failure: Arc<Error>) {
289        let mut previous = None;
290        self.status.send_modify(|status| {
291            if !status.phase.is_terminal() {
292                status.outcome.get_or_insert(Outcome::Unavailable);
293                status.cleanup_pending = true;
294                previous = status.cleanup_failure.replace(failure);
295            }
296        });
297        drop(previous);
298    }
299
300    pub(crate) fn is_settled(&self) -> bool {
301        self.status.borrow().phase.is_terminal()
302    }
303
304    pub(crate) fn track(&self, child: &Arc<Self>) {
305        let mut children = self.children.lock().expect("instance child lock is healthy");
306        children.retain(|child| child.upgrade().is_some_and(|child| !child.is_settled()));
307        children.push(Arc::downgrade(child));
308        drop(children);
309        if self.status().cancellation_requested {
310            child.cancel();
311        }
312    }
313
314    #[cfg(any(feature = "plugin-sidecar", test))]
315    pub(crate) fn depend_on(self: &Arc<Self>, provider: &Arc<Self>, retirement: bool) -> crate::Result<()> {
316        if Arc::ptr_eq(self, provider) {
317            return Err(Error::InvalidInput { field: "physical provider", reason: "an instance cannot retain itself" });
318        }
319        let mut providers = self.providers.lock().expect("instance providers are healthy");
320        if providers.iter().any(|held| Arc::ptr_eq(&held.provider, provider)) {
321            return Ok(());
322        }
323        let mut admitted = false;
324        let mut stopping_control = false;
325        provider.status.send_modify(|status| {
326            if status.phase.is_terminal() {
327                return;
328            }
329            if status.cancellation_requested {
330                if !retirement {
331                    return;
332                }
333                if status.dependents == 0 {
334                    // The Existing Physical Stop Owner Drives Its Own SDK Controls.
335                    stopping_control = true;
336                    return;
337                }
338            }
339            if let Some(count) = status.dependents.checked_add(1) {
340                status.dependents = count;
341                admitted = true;
342            }
343        });
344        if stopping_control {
345            return Ok(());
346        }
347        if !admitted {
348            return Err(Error::Closing { domain: "physical provider" });
349        }
350        providers.push(Dependency { provider: provider.clone() });
351        drop(providers);
352        if !retirement {
353            provider.track(self);
354        }
355        Ok(())
356    }
357
358    pub(crate) fn retain<R: Resource>(&self, resource: R, failure: Arc<Error>) {
359        let mut retained = self.retained.lock().expect("instance cleanup lock is healthy");
360        assert!(retained.is_none(), "one instance has one physical cleanup owner");
361        *retained = Some(Arc::new(Cleanup::new(resource)));
362        self.status.send_modify(|status| {
363            if status.phase != Phase::Stopping {
364                status.phase = Phase::Finalizing;
365            }
366            status.cleanup_pending = true;
367            status.cleanup_failure = Some(failure);
368        });
369    }
370
371    pub(crate) fn settle_dependents(&self) -> Finalize<'_> {
372        Box::pin(async move {
373            let children: Vec<_> = self.children.lock().expect("instance child lock is healthy").iter().filter_map(Weak::upgrade).collect();
374            let pending = children.into_iter().map(|child| async move { child.settle().await });
375            super::settle_all(pending).await
376        })
377    }
378
379    pub(crate) fn settle(&self) -> Finalize<'_> {
380        Box::pin(async move {
381            let mut changed = self.settlement();
382            loop {
383                changed.borrow_and_update();
384                self.settle_dependents().await?;
385                let retained = self.retained.lock().expect("instance cleanup lock is healthy").clone();
386                if let Some(retained) = retained {
387                    if let Err(failure) = retained.settle().await {
388                        let mut previous = None;
389                        self.status.send_modify(|status| {
390                            if !status.phase.is_terminal() {
391                                previous = status.cleanup_failure.replace(Arc::clone(&failure));
392                            }
393                        });
394                        drop(previous);
395                        return Err(failure);
396                    }
397                    self.finished();
398                    return Ok(());
399                }
400                let status = self.status();
401                if status.phase.is_terminal() {
402                    return Ok(());
403                }
404                if let Some(failure) = status.cleanup_failure {
405                    return Err(failure);
406                }
407                if changed.changed().await.is_err() {
408                    return Err(Arc::new(Error::Closing { domain: "instance settlement" }));
409                }
410            }
411        })
412    }
413
414    pub(crate) fn cancel(&self) {
415        let mut interrupt = false;
416        self.status.send_modify(|status| {
417            interrupt = status.dependents == 0;
418            status.cancellation_requested = true;
419            if !status.phase.is_terminal() {
420                status.cleanup_pending = true;
421            }
422            if !matches!(status.phase, Phase::Finalizing) && !status.phase.is_terminal() {
423                status.phase = Phase::Stopping;
424            }
425        });
426        if interrupt {
427            self.cancellation.send_replace(true);
428        }
429        let children: Vec<_> = self.children.lock().expect("instance child lock is healthy").iter().filter_map(Weak::upgrade).collect();
430        for child in children {
431            child.cancel();
432        }
433    }
434}
435
436impl<R: Resource> Settlement for Cleanup<R> {
437    fn settle(&self) -> Finalize<'_> {
438        Box::pin(self.settle())
439    }
440}
441
442impl Drop for Dependency {
443    fn drop(&mut self) {
444        let mut interrupt = false;
445        self.provider.status.send_modify(|status| {
446            status.dependents = status.dependents.checked_sub(1).expect("each provider hold releases once");
447            interrupt = status.cancellation_requested && status.dependents == 0;
448        });
449        if interrupt {
450            self.provider.cancellation.send_replace(true);
451        }
452    }
453}
454
455/* === Public Functions === */
456
457/* === Private Functions === */
458
459/* === Tests === */
harmos_runtime/supervision/instance.rs

harmos_runtime/supervision/
instance.rs
