appcore_supervisor/
supervisor.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: supervisor.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/24 11:51:10 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 13:18:47 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Process-local managed-service orchestration.
12
13use crate::graph::{topological_order, validate_dependencies};
14use crate::restart_executor::{
15    RestartCommand, RestartCompletion, RestartExecutor, RestartOutcome,
16    DEFAULT_RESTART_QUEUE_CAPACITY, DEFAULT_RESTART_WORKERS,
17};
18use crate::{
19    DependencyRequirement, ManagedService, RestartMode, RestartState, ServiceHealth,
20    ServiceRuntimeState, SupervisorDiagnosis, SupervisorError, SupervisorEvent,
21    SupervisorEventKind, SupervisorResult, SupervisorWatchdog, WatchdogConfig, WatchdogSnapshot,
22    WatchdogState, DEFAULT_EVENT_CAPACITY,
23};
24use std::collections::{BTreeMap, VecDeque};
25use std::sync::atomic::{AtomicU64, Ordering};
26use std::sync::{Arc, Mutex, RwLock};
27use std::time::{Duration, SystemTime, UNIX_EPOCH};
28
29#[path = "supervisor_diagnostics.rs"]
30mod diagnostics;
31#[path = "supervisor_lifecycle.rs"]
32mod lifecycle;
33#[path = "supervisor_restart.rs"]
34mod restart;
35
36pub(super) struct RuntimeRecord {
37    health: Option<ServiceHealth>,
38    runtime_state: ServiceRuntimeState,
39    restart_state: RestartState,
40    restart_times_ms: VecDeque<u64>,
41    restart_count: u64,
42    operator_required: bool,
43    quarantined: bool,
44}
45
46impl Default for RuntimeRecord {
47    fn default() -> Self {
48        Self {
49            health: None,
50            runtime_state: ServiceRuntimeState::Stopped,
51            restart_state: RestartState::None,
52            restart_times_ms: VecDeque::new(),
53            restart_count: 0,
54            operator_required: false,
55            quarantined: false,
56        }
57    }
58}
59
60struct SupervisorInner {
61    services: RwLock<BTreeMap<String, Arc<dyn ManagedService>>>,
62    records: Mutex<BTreeMap<String, RuntimeRecord>>,
63    events: Mutex<VecDeque<SupervisorEvent>>,
64    event_capacity: usize,
65    event_sequence: AtomicU64,
66    jitter_state: AtomicU64,
67    watchdog: Arc<SupervisorWatchdog>,
68    restart_executor: RestartExecutor,
69}
70
71/// Process-local orchestrator for Runtime-owned managed services.
72///
73/// The supervisor never starts, stops, or replaces its host process.
74#[derive(Clone)]
75pub struct Supervisor {
76    inner: Arc<SupervisorInner>,
77}
78
79impl Supervisor {
80    /// Creates an empty supervisor with safe watchdog and executor defaults.
81    pub fn new() -> Self {
82        let created_at_ms = now_ms();
83        Self::assemble(
84            DEFAULT_EVENT_CAPACITY,
85            created_at_ms,
86            SupervisorWatchdog::with_default(created_at_ms),
87        )
88    }
89
90    /// Creates an empty supervisor with installation watchdog policy.
91    pub fn with_watchdog_config(config: WatchdogConfig) -> SupervisorResult<Self> {
92        Self::with_options(DEFAULT_EVENT_CAPACITY, config, now_ms())
93    }
94
95    /// Creates an empty supervisor with an explicit event bound.
96    pub fn with_event_capacity(event_capacity: usize) -> Self {
97        let created_at_ms = now_ms();
98        Self::assemble(
99            event_capacity,
100            created_at_ms,
101            SupervisorWatchdog::with_default(created_at_ms),
102        )
103    }
104
105    fn with_options(
106        event_capacity: usize,
107        watchdog_config: WatchdogConfig,
108        created_at_ms: u64,
109    ) -> SupervisorResult<Self> {
110        let watchdog = SupervisorWatchdog::new(watchdog_config, created_at_ms)?;
111        Ok(Self::assemble(event_capacity, created_at_ms, watchdog))
112    }
113
114    fn assemble(event_capacity: usize, created_at_ms: u64, watchdog: SupervisorWatchdog) -> Self {
115        Self {
116            inner: Arc::new(SupervisorInner {
117                services: RwLock::new(BTreeMap::new()),
118                records: Mutex::new(BTreeMap::new()),
119                events: Mutex::new(VecDeque::new()),
120                event_capacity: event_capacity.max(1),
121                event_sequence: AtomicU64::new(0),
122                jitter_state: AtomicU64::new(created_at_ms.max(1)),
123                watchdog: Arc::new(watchdog),
124                restart_executor: RestartExecutor::new(
125                    DEFAULT_RESTART_QUEUE_CAPACITY,
126                    DEFAULT_RESTART_WORKERS,
127                ),
128            }),
129        }
130    }
131
132    /// Reports whether two handles reference the same supervisor instance.
133    pub fn same_instance(&self, other: &Self) -> bool {
134        Arc::ptr_eq(&self.inner, &other.inner)
135    }
136
137    /// Returns the lock-independent watchdog shared with health consumers.
138    pub fn watchdog(&self) -> Arc<SupervisorWatchdog> {
139        Arc::clone(&self.inner.watchdog)
140    }
141
142    /// Registers one unique service without starting it.
143    pub fn register(&self, service: Arc<dyn ManagedService>) -> SupervisorResult<()> {
144        service.descriptor().validate()?;
145        let name = service.descriptor().name().to_string();
146        let mut services = self
147            .inner
148            .services
149            .write()
150            .map_err(|_| SupervisorError::StatePoisoned)?;
151        if services.contains_key(&name) {
152            return Err(SupervisorError::ServiceAlreadyRegistered(name));
153        }
154        services.insert(name.clone(), service);
155        self.inner
156            .records
157            .lock()
158            .map_err(|_| SupervisorError::StatePoisoned)?
159            .insert(name, RuntimeRecord::default());
160        Ok(())
161    }
162
163    /// Registers a service or replaces an inactive deployment placeholder.
164    ///
165    /// An enabled registration is never replaced because it may own resources.
166    pub fn register_or_replace_inactive(
167        &self,
168        service: Arc<dyn ManagedService>,
169    ) -> SupervisorResult<()> {
170        service.descriptor().validate()?;
171        let name = service.descriptor().name().to_string();
172        let mut services = self
173            .inner
174            .services
175            .write()
176            .map_err(|_| SupervisorError::StatePoisoned)?;
177        if services
178            .get(&name)
179            .is_some_and(|current| current.descriptor().activation().is_enabled())
180        {
181            return Err(SupervisorError::ServiceAlreadyRegistered(name));
182        }
183        services.insert(name.clone(), service);
184        self.inner
185            .records
186            .lock()
187            .map_err(|_| SupervisorError::StatePoisoned)?
188            .insert(name, RuntimeRecord::default());
189        Ok(())
190    }
191
192    /// Validates policies and returns dependency-first service order.
193    pub fn validate(&self) -> SupervisorResult<Vec<String>> {
194        let services = self
195            .inner
196            .services
197            .read()
198            .map_err(|_| SupervisorError::StatePoisoned)?;
199        validate_dependencies(&services)?;
200        topological_order(&services)
201    }
202
203    /// Starts all enabled services in dependency order.
204    pub fn start_all(&self) -> SupervisorResult<()> {
205        for name in self.validate()? {
206            let service = self.service(&name)?;
207            if service.descriptor().activation().is_enabled() {
208                self.start(&name, now_ms())?;
209            }
210        }
211        Ok(())
212    }
213
214    /// Starts one enabled service after checking all dependencies.
215    pub fn start(&self, name: &str, timestamp_ms: u64) -> SupervisorResult<()> {
216        let service = self.service(name)?;
217        if !service.descriptor().activation().is_enabled() {
218            return Ok(());
219        }
220        self.require_dependencies(&service)?;
221        let previous = self.record_health(name)?;
222        match service.start() {
223            Ok(()) => {
224                let health = service.health();
225                self.update_record(name, health, service.runtime_state())?;
226                self.emit(
227                    name,
228                    SupervisorEventKind::ServiceStarted,
229                    timestamp_ms,
230                    0,
231                    states(previous, health),
232                    "lifecycle_start",
233                );
234                Ok(())
235            }
236            Err(error) => {
237                self.update_record(name, ServiceHealth::Failed, service.runtime_state())?;
238                self.emit(
239                    name,
240                    SupervisorEventKind::ServiceFailed,
241                    timestamp_ms,
242                    0,
243                    states(previous, ServiceHealth::Failed),
244                    "start_failed",
245                );
246                Err(error)
247            }
248        }
249    }
250
251    pub(super) fn service(&self, name: &str) -> SupervisorResult<Arc<dyn ManagedService>> {
252        self.inner
253            .services
254            .read()
255            .map_err(|_| SupervisorError::StatePoisoned)?
256            .get(name)
257            .cloned()
258            .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))
259    }
260
261    pub(super) fn records(
262        &self,
263    ) -> SupervisorResult<std::sync::MutexGuard<'_, BTreeMap<String, RuntimeRecord>>> {
264        self.inner
265            .records
266            .lock()
267            .map_err(|_| SupervisorError::StatePoisoned)
268    }
269
270    pub(super) fn update_record(
271        &self,
272        name: &str,
273        health: ServiceHealth,
274        runtime_state: ServiceRuntimeState,
275    ) -> SupervisorResult<Option<ServiceHealth>> {
276        let mut records = self.records()?;
277        let record = record_mut(&mut records, name)?;
278        record.runtime_state = runtime_state;
279        Ok(record.health.replace(health))
280    }
281
282    fn record_health(&self, name: &str) -> SupervisorResult<Option<ServiceHealth>> {
283        let records = self.records()?;
284        records
285            .get(name)
286            .map(|record| record.health)
287            .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))
288    }
289
290    pub(super) fn restart_attempt(&self, name: &str) -> u64 {
291        self.inner
292            .records
293            .lock()
294            .ok()
295            .and_then(|records| records.get(name).map(|record| record.restart_count))
296            .unwrap_or(0)
297    }
298
299    pub(super) fn emit(
300        &self,
301        service: &str,
302        kind: SupervisorEventKind,
303        timestamp_ms: u64,
304        attempt: u64,
305        transition: (&str, &str),
306        reason: &str,
307    ) {
308        let trace = self
309            .inner
310            .event_sequence
311            .fetch_add(1, Ordering::AcqRel)
312            .saturating_add(1);
313        let Ok(mut events) = self.inner.events.lock() else {
314            return;
315        };
316        while events.len() >= self.inner.event_capacity {
317            events.pop_front();
318        }
319        events.push_back(SupervisorEvent::new(
320            service,
321            kind,
322            timestamp_ms,
323            attempt,
324            transition,
325            reason,
326            format!("supervisor-{trace}"),
327        ));
328    }
329
330    pub(super) fn jitter(&self, maximum: Duration) -> Duration {
331        let maximum_ms = duration_ms(maximum);
332        if maximum_ms == 0 {
333            return Duration::ZERO;
334        }
335        let mut current = self.inner.jitter_state.load(Ordering::Relaxed);
336        loop {
337            let mut next = current;
338            next ^= next << 13;
339            next ^= next >> 7;
340            next ^= next << 17;
341            match self.inner.jitter_state.compare_exchange_weak(
342                current,
343                next,
344                Ordering::Relaxed,
345                Ordering::Relaxed,
346            ) {
347                Ok(_) => return Duration::from_millis(next % maximum_ms.saturating_add(1)),
348                Err(observed) => current = observed,
349            }
350        }
351    }
352}
353
354impl Default for Supervisor {
355    fn default() -> Self {
356        Self::new()
357    }
358}
359
360pub(super) fn record_mut<'a>(
361    records: &'a mut BTreeMap<String, RuntimeRecord>,
362    name: &str,
363) -> SupervisorResult<&'a mut RuntimeRecord> {
364    records
365        .get_mut(name)
366        .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))
367}
368
369pub(super) fn states(
370    previous: Option<ServiceHealth>,
371    next: ServiceHealth,
372) -> (&'static str, &'static str) {
373    (
374        previous.map(health_name).unwrap_or("Unknown"),
375        health_name(next),
376    )
377}
378
379pub(super) fn health_name(health: ServiceHealth) -> &'static str {
380    match health {
381        ServiceHealth::Ready => "Ready",
382        ServiceHealth::Healthy => "Healthy",
383        ServiceHealth::Degraded => "Degraded",
384        ServiceHealth::Failed => "Failed",
385        ServiceHealth::Starting => "Starting",
386        ServiceHealth::Stopping => "Stopping",
387        ServiceHealth::Unknown => "Unknown",
388    }
389}
390
391pub(super) fn watchdog_states(
392    previous: WatchdogState,
393    next: WatchdogState,
394) -> (&'static str, &'static str) {
395    (watchdog_name(previous), watchdog_name(next))
396}
397
398fn watchdog_name(state: WatchdogState) -> &'static str {
399    match state {
400        WatchdogState::Starting => "Starting",
401        WatchdogState::Healthy => "Healthy",
402        WatchdogState::Stalled => "Stalled",
403        WatchdogState::Failed => "Failed",
404        WatchdogState::Stopping => "Stopping",
405    }
406}
407
408pub(super) fn duration_ms(duration: Duration) -> u64 {
409    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
410}
411
412pub(super) fn now_ms() -> u64 {
413    SystemTime::now()
414        .duration_since(UNIX_EPOCH)
415        .map(|duration| duration.as_millis() as u64)
416        .unwrap_or(0)
417}
418
419#[cfg(test)]
420#[path = "supervisor_tests.rs"]
421mod tests;
appcore_supervisor/supervisor.rs

appcore_supervisor/
supervisor.rs
