stackless_daemon/
adopt.rs

1//! Re-adoption on daemon start (ARCHITECTURE.md §3: upgrade = restart +
2//! re-adopt). Routes and supervision records live only in the daemon's
3//! memory, so they die with the daemon — re-adoption rebuilds them from
4//! the checkpoint journal, the one durable truth. For every ACTIVE
5//! instance, each `start:*` checkpoint whose recorded process is still
6//! alive has its proxy routes and supervision record re-registered; a
7//! dead process is noted, not restarted (v0 supervision policy: observe,
8//! don't restart).
9
10use std::sync::Arc;
11
12use stackless_core::checkpoint::StartCheckpoint;
13use stackless_core::paths::Paths;
14use stackless_core::process::ProcessStamp;
15use stackless_core::state::{InstanceStatus, Store};
16use stackless_core::types::DnsName;
17
18use crate::state::DaemonState;
19
20/// What one adoption pass observed — for the daemon log only.
21#[derive(Debug, Default)]
22pub struct AdoptionSummary {
23    pub adopted: Vec<String>,
24    pub dead: Vec<String>,
25}
26
27/// Rebuild routing and supervision from the journal. Opens the store
28/// fresh (short-lived access; the store is multi-process-safe). Errors
29/// reading the store are non-fatal: re-adoption is best-effort recovery,
30/// and the daemon must come up regardless.
31pub fn readopt(state: &Arc<DaemonState>, paths: &Paths) -> AdoptionSummary {
32    let mut summary = AdoptionSummary::default();
33    let store = match Store::open_with_paths(paths) {
34        Ok(store) => store,
35        Err(_) => return summary,
36    };
37    let instances = match store.instances() {
38        Ok(instances) => instances,
39        Err(_) => return summary,
40    };
41    for record in instances {
42        if record.status != InstanceStatus::Active {
43            continue;
44        }
45        let checkpoints = match store.checkpoints(record.name.as_str()) {
46            Ok(checkpoints) => checkpoints,
47            Err(_) => continue,
48        };
49        for checkpoint in checkpoints {
50            let Some(service) = checkpoint.step_id.strip_prefix("start:") else {
51                continue;
52            };
53            let Ok(start) = serde_json::from_str::<StartCheckpoint>(&checkpoint.payload) else {
54                continue;
55            };
56            let Ok(service_name) = DnsName::try_new(service) else {
57                continue;
58            };
59            let stamp = ProcessStamp {
60                pid: start.pid,
61                start_time: start.start_time,
62            };
63            let label = format!("{}/{service}", record.name.as_str());
64            if !stamp.is_alive() {
65                summary.dead.push(label);
66                continue;
67            }
68            for host in &start.hosts {
69                state.route_set(host.clone(), start.port);
70            }
71            state.supervise(record.name.clone(), service_name, stamp);
72            summary.adopted.push(label);
73        }
74    }
75    summary
76}
stackless_daemon/adopt.rs

stackless_daemon/
adopt.rs
