frame_core/registry/
recovery.rs

1use std::sync::Arc;
2
3use beamr::process::ExitReason;
4use beamr::scheduler::Scheduler;
5
6use super::{ComponentId, ComponentRecord, ComponentRegistry, RegistryError};
7use crate::error::TombstoneKind;
8use crate::supervision::observation::{ExitObservation, ObservationError};
9
10/// Honest account of the host-forced cleanup used to recover a Failed tree.
11#[derive(Clone, Debug, Eq, PartialEq)]
12pub struct ForcedCleanupReport {
13    /// Disposition of every child still retained after the failed drain.
14    pub children: Vec<ChildCleanup>,
15    /// Disposition of the retained native supervisor, when one existed.
16    pub supervisor: Option<ProcessCleanup>,
17    /// Monitor failure observed while joining it, when it did not exit cleanly.
18    pub monitor_failure: Option<String>,
19}
20
21/// Cleanup result for one named child process.
22#[derive(Clone, Debug, Eq, PartialEq)]
23pub struct ChildCleanup {
24    /// Child key from its retained declaration.
25    pub name: String,
26    /// Process identifier whose tombstone was observed.
27    pub pid: u64,
28    /// Whether Frame killed a survivor or found it already dead.
29    pub disposition: ProcessCleanup,
30}
31
32/// How one process reached a tombstone during failed-tree recovery.
33#[derive(Clone, Copy, Debug, Eq, PartialEq)]
34pub enum ProcessCleanup {
35    /// The process was alive, so Frame terminated it and observed this tombstone.
36    Killed {
37        /// Tombstone observed after host termination.
38        tombstone: TombstoneKind,
39    },
40    /// The process had already exited; Frame observed its existing tombstone.
41    AlreadyDead {
42        /// Existing tombstone observed without termination.
43        tombstone: TombstoneKind,
44    },
45}
46
47impl ComponentRegistry {
48    pub(super) fn recover_failed_tree(
49        &self,
50        id: ComponentId,
51        record: &Arc<ComponentRecord>,
52    ) -> Result<Option<ForcedCleanupReport>, RegistryError> {
53        let monitor_failure = record
54            .tree
55            .lock()
56            .map_err(|_| RegistryError::SynchronizationPoisoned)?
57            .take()
58            .and_then(|tree| tree.join_finished(id).err())
59            .map(|error| error.to_string());
60        let children = record
61            .status
62            .children
63            .lock()
64            .map_err(|_| RegistryError::SynchronizationPoisoned)?
65            .clone();
66        let mut cleaned_children = Vec::with_capacity(children.len());
67        for child in children {
68            cleaned_children.push(ChildCleanup {
69                name: child.name,
70                pid: child.pid,
71                disposition: cleanup_process(id, &self.scheduler, child.pid, self.config)?,
72            });
73        }
74        record
75            .status
76            .children
77            .lock()
78            .map_err(|_| RegistryError::SynchronizationPoisoned)?
79            .clear();
80        let supervisor_pid = *record
81            .status
82            .supervisor
83            .lock()
84            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
85        let supervisor = supervisor_pid
86            .map(|pid| cleanup_process(id, &self.scheduler, pid, self.config))
87            .transpose()?;
88        if supervisor.is_some() {
89            *record
90                .status
91                .supervisor
92                .lock()
93                .map_err(|_| RegistryError::SynchronizationPoisoned)? = None;
94        }
95        if cleaned_children.is_empty() && supervisor.is_none() && monitor_failure.is_none() {
96            Ok(None)
97        } else {
98            Ok(Some(ForcedCleanupReport {
99                children: cleaned_children,
100                supervisor,
101                monitor_failure,
102            }))
103        }
104    }
105}
106
107fn cleanup_process(
108    id: ComponentId,
109    scheduler: &Arc<Scheduler>,
110    pid: u64,
111    config: crate::supervision::LifecycleConfig,
112) -> Result<ProcessCleanup, RegistryError> {
113    let observation = ExitObservation::register(scheduler, pid, config.operation_timeout)
114        .map_err(|error| recovery_observation_error(id, pid, error))?;
115    let was_alive = scheduler.process_table().get(pid).is_some();
116    if was_alive {
117        scheduler.terminate_process(pid, ExitReason::Kill);
118    }
119    let reason = observation
120        .wait()
121        .map_err(|error| recovery_observation_error(id, pid, error))?;
122    let tombstone = tombstone_kind(reason);
123    if was_alive {
124        Ok(ProcessCleanup::Killed { tombstone })
125    } else {
126        Ok(ProcessCleanup::AlreadyDead { tombstone })
127    }
128}
129
130fn recovery_observation_error(id: ComponentId, pid: u64, error: ObservationError) -> RegistryError {
131    let detail = match error {
132        ObservationError::Spawn(detail) => format!("recovery observer spawn failed: {detail}"),
133        ObservationError::ReadyDeadline => "recovery observer readiness timed out".to_owned(),
134        ObservationError::Registration(detail) => {
135            format!("recovery monitor registration failed: {detail}")
136        }
137        ObservationError::TombstoneDeadline => {
138            format!("timed out observing recovered process {pid} tombstone")
139        }
140        ObservationError::Protocol(detail) => detail,
141        ObservationError::CleanupDeadline => "recovery observer cleanup timed out".to_owned(),
142    };
143    RegistryError::SupervisorProtocol { id, detail }
144}
145
146const fn tombstone_kind(reason: ExitReason) -> TombstoneKind {
147    match reason {
148        ExitReason::Normal => TombstoneKind::Normal,
149        ExitReason::Kill => TombstoneKind::Kill,
150        ExitReason::Killed => TombstoneKind::Killed,
151        ExitReason::Error => TombstoneKind::Error,
152        ExitReason::NoConnection => TombstoneKind::NoConnection,
153        ExitReason::NoProc => TombstoneKind::NoProcess,
154    }
155}
frame_core/registry/recovery.rs

frame_core/registry/
recovery.rs
