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

harmos_runtime/supervision/
instance.rs
