tawasuyu/brahman

1//! sandokan-local — `LocalEngine`, orquestación in-process.
2//!
3//! Primera implementación del trait [`Engine`]: encarna Cards en el
4//! mismo host vía `arje-incarnate`, mantiene un registro de las
5//! entidades activas y hace reaping perezoso (waitpid WNOHANG en cada
6//! consulta) para detectar salidas sin un task de fondo dedicado.
7//!
8//! `DaemonEngine` y `RemoteEngine` (transportes) se construirán sobre
9//! este mismo contrato en crates separados.
10
11mod proc;
12
13use arje_incarnate::{Incarnator, IncarnatorConfig};
14use async_trait::async_trait;
15use nix::sys::signal::{kill, Signal};
16use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
17use nix::unistd::Pid;
18use sandokan_core::{Engine, EngineError, ExecHandle, Intent, TelemetryFrame};
19use sandokan_lifecycle::LifecycleState;
20use std::collections::HashMap;
21use std::sync::Mutex;
22use std::time::{Duration, Instant, SystemTime};
23use ulid::Ulid;
24
25/// Una entidad encarnada y su estado conocido.
26struct Entity {
27    handle: ExecHandle,
28    pid: i32,
29    state: LifecycleState,
30}
31
32/// Orquestador in-process. Encarna Cards localmente y trackea su lifecycle.
33pub struct LocalEngine {
34    base_cfg: IncarnatorConfig,
35    registry: Mutex<HashMap<Ulid, Entity>>,
36}
37
38impl LocalEngine {
39    /// Crea un engine con configuración de incarnación por defecto.
40    pub fn new() -> Self {
41        Self::with_config(IncarnatorConfig::default())
42    }
43
44    /// Crea un engine con una `IncarnatorConfig` explícita (bus socket,
45    /// env extra, strict_caps).
46    pub fn with_config(cfg: IncarnatorConfig) -> Self {
47        Self {
48            base_cfg: cfg,
49            registry: Mutex::new(HashMap::new()),
50        }
51    }
52
53    /// Marca el estado de una entidad (best-effort; ignora lock envenenado).
54    fn mark(&self, card_id: Ulid, state: LifecycleState) {
55        if let Ok(mut reg) = self.registry.lock() {
56            if let Some(ent) = reg.get_mut(&card_id) {
57                ent.state = state;
58            }
59        }
60    }
61}
62
63impl Default for LocalEngine {
64    fn default() -> Self {
65        Self::new()
66    }
67}
68
69/// Reaping no-bloqueante de un pid. `Some(estado terminal)` si la entidad
70/// transicionó; `None` si sigue viva.
71fn reap(pid: i32) -> Option<LifecycleState> {
72    match waitpid(Pid::from_raw(pid), Some(WaitPidFlag::WNOHANG)) {
73        Ok(WaitStatus::Exited(_, code)) => Some(LifecycleState::Exited { code }),
74        Ok(WaitStatus::Signaled(_, _, _)) => Some(LifecycleState::Killed),
75        Ok(WaitStatus::StillAlive) => None,
76        // Stopped / Continued / PtraceEvent: la entidad sigue presente.
77        Ok(_) => None,
78        Err(nix::errno::Errno::ECHILD) => {
79            // No es (ya) hijo reapable. Si procfs no lo tiene, terminó.
80            if proc::proc_exists(pid) {
81                None
82            } else {
83                Some(LifecycleState::Exited { code: -1 })
84            }
85        }
86        Err(_) => None,
87    }
88}
89
90#[async_trait]
91impl Engine for LocalEngine {
92    async fn run(&self, intent: Intent) -> Result<ExecHandle, EngineError> {
93        let card_id = intent.card_id();
94        let label = intent.card.label.clone();
95
96        // El env del contexto se mergea sobre el del engine base.
97        let mut cfg = self.base_cfg.clone();
98        cfg.extra_env.extend(intent.context.env.clone());
99
100        // NOTA v1: `IsolationLevel` es advisory. None/Standard encarnan
101        // según `Card.soma`; Sealed (rootfs aislado vía pivot_root +
102        // OverlayFS) queda reservado para cuando el Intent transporte una
103        // spec de rootfs — `arje-incarnate` ya expone los builders.
104        let incarnator = Incarnator::new(cfg);
105        let outcome = incarnator
106            .incarnate(&intent.card)
107            .map_err(|e| EngineError::Incarnate(e.to_string()))?;
108
109        let handle = ExecHandle {
110            card_id,
111            label,
112            started_at: SystemTime::now(),
113        };
114        let mut reg = self.registry.lock().expect("registry lock");
115        reg.insert(
116            card_id,
117            Entity {
118                handle: handle.clone(),
119                pid: outcome.pid.as_raw(),
120                state: LifecycleState::Running,
121            },
122        );
123        Ok(handle)
124    }
125
126    async fn stop(&self, card_id: Ulid, grace: Duration) -> Result<(), EngineError> {
127        let pid = {
128            let reg = self.registry.lock().expect("registry lock");
129            reg.get(&card_id)
130                .map(|e| e.pid)
131                .ok_or(EngineError::NotFound(card_id))?
132        };
133        let npid = Pid::from_raw(pid);
134
135        // SIGTERM + período de gracia: damos chance a un cierre ordenado.
136        if grace > Duration::ZERO {
137            let _ = kill(npid, Signal::SIGTERM);
138            let deadline = Instant::now() + grace;
139            loop {
140                if reap(pid).is_some() {
141                    self.mark(card_id, LifecycleState::Killed);
142                    return Ok(());
143                }
144                if Instant::now() >= deadline {
145                    break;
146                }
147                tokio::time::sleep(Duration::from_millis(20)).await;
148            }
149        }
150
151        // SIGKILL + reap bloqueante de lo que haya quedado.
152        let _ = kill(npid, Signal::SIGKILL);
153        let _ = waitpid(npid, None);
154        self.mark(card_id, LifecycleState::Killed);
155        Ok(())
156    }
157
158    async fn list(&self) -> Result<Vec<ExecHandle>, EngineError> {
159        let mut reg = self.registry.lock().expect("registry lock");
160        let mut out = Vec::new();
161        for ent in reg.values_mut() {
162            if !ent.state.is_terminal() {
163                if let Some(new_state) = reap(ent.pid) {
164                    ent.state = new_state;
165                }
166            }
167            if !ent.state.is_terminal() {
168                out.push(ent.handle.clone());
169            }
170        }
171        Ok(out)
172    }
173
174    async fn status(&self, card_id: Ulid) -> Result<LifecycleState, EngineError> {
175        let mut reg = self.registry.lock().expect("registry lock");
176        let ent = reg
177            .get_mut(&card_id)
178            .ok_or(EngineError::NotFound(card_id))?;
179        if !ent.state.is_terminal() {
180            if let Some(new_state) = reap(ent.pid) {
181                ent.state = new_state;
182            }
183        }
184        Ok(ent.state.clone())
185    }
186
187    async fn telemetry(&self, card_id: Ulid) -> Result<TelemetryFrame, EngineError> {
188        let pid = {
189            let reg = self.registry.lock().expect("registry lock");
190            reg.get(&card_id)
191                .map(|e| e.pid)
192                .ok_or(EngineError::NotFound(card_id))?
193        };
194        Ok(TelemetryFrame {
195            card_id,
196            at: SystemTime::now(),
197            mem_bytes: proc::read_mem_bytes(pid),
198            nproc: proc::read_thread_count(pid),
199            // v1: CPU% requiere dos samples espaciados — pendiente.
200            cpu_pct: 0.0,
201        })
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use super::*;
208
209    #[tokio::test]
210    async fn empty_engine_lists_nothing() {
211        let e = LocalEngine::new();
212        assert!(e.list().await.unwrap().is_empty());
213    }
214
215    #[tokio::test]
216    async fn unknown_card_is_not_found() {
217        let e = LocalEngine::new();
218        let id = Ulid::new();
219        assert!(matches!(e.status(id).await, Err(EngineError::NotFound(_))));
220        assert!(matches!(
221            e.stop(id, Duration::ZERO).await,
222            Err(EngineError::NotFound(_))
223        ));
224        assert!(matches!(
225            e.telemetry(id).await,
226            Err(EngineError::NotFound(_))
227        ));
228    }
229}