tawasuyu/brahman

1//! Encarnación, muerte y supervisión.
2//!
3//! Aquí vive el flujo: Card → autorizar → soma::incarnate / wasm → registro
4//! en el grafo → SIGCHLD → on_death → Restart/OneShot/Delegate.
5
6use super::{EnteGraph, Incarnated};
7use crate::events::{ExitStatus, GraphEvent};
8use ente_bus::{BusMessage, BusPayload, BusRequest};
9use ente_card::{Capability, EntityCard, Payload, Supervision};
10use tokio::sync::mpsc;
11use tracing::{info, warn};
12use ulid::Ulid;
13
14impl EnteGraph {
15    /// Encarna las dependencias declaradas en la Semilla. Único punto donde
16    /// el Init "decide": después sólo reacciona.
17    pub async fn instantiate_seed_dependencies(
18        &mut self,
19        _tx: &mpsc::Sender<GraphEvent>,
20    ) -> anyhow::Result<()> {
21        let cards = std::mem::take(&mut self.pending_genesis);
22        if cards.is_empty() {
23            info!(seed = %self.seed.label, "semilla sin genesis cards");
24            return Ok(());
25        }
26        info!(seed = %self.seed.label, count = cards.len(), "instanciando genesis");
27        let seed_id = self.seed.id;
28        for card in cards {
29            if let Err(e) = self.authorize_and_spawn(card, seed_id).await {
30                warn!(?e, "genesis card falló");
31            }
32        }
33        Ok(())
34    }
35
36    /// Spawn solicitado por un Ente con `Capability::Spawn`. Verifica auth,
37    /// requires del grafo, y delega la encarnación al backend correspondiente
38    /// (`ente_soma` para procesos, `ente_wasm` para Wasm).
39    pub async fn authorize_and_spawn(
40        &mut self,
41        mut card: EntityCard,
42        requester: Ulid,
43    ) -> anyhow::Result<()> {
44        if !self.holder_has(requester, &Capability::Spawn) {
45            warn!(?requester, "spawn denied: lacks Capability::Spawn");
46            return Ok(());
47        }
48        if let Err(e) = card.validate() {
49            warn!(?e, label = %card.label, "card inválida, spawn rechazado");
50            return Ok(());
51        }
52        // Falla rápida sobre `requires` — mejor que daemons en bucle.
53        for req in &card.requires {
54            if !self.providers.contains_key(req) {
55                warn!(?req, label = %card.label, "requires no satisfecho");
56                return Ok(());
57            }
58        }
59        // Lineage por defecto = quien pidió el spawn.
60        if card.lineage.is_none() {
61            card.lineage = Some(requester);
62        }
63
64        let pid = match &card.payload {
65            Payload::Virtual => None,
66            Payload::Native { .. } | Payload::Legacy { .. } => {
67                Some(ente_soma::incarnate(&card)?)
68            }
69            Payload::Wasm { module_sha256, entry } => {
70                // Wasm: hilo dedicado, sin PID. Su muerte se observa por
71                // estado del runtime, no por SIGCHLD.
72                let bytes = ente_cas::resolve(module_sha256)
73                    .map_err(|e| anyhow::anyhow!("CAS resolve para {}: {e}", card.label))?;
74                ente_wasm::incarnate_wasm(&card, bytes, entry.clone())?;
75                None
76            }
77        };
78
79        if let Some(p) = pid {
80            self.by_pid.insert(p.as_raw(), card.id);
81        }
82        self.register_provider(&card);
83        if let Some(parent) = card.lineage {
84            self.children.entry(parent).or_default().push(card.id);
85        }
86        info!(label = %card.label, ?pid, lineage = ?card.lineage, "Ente encarnado");
87        self.incarnated.insert(card.id, Incarnated {
88            card, pid,
89            dynamic_provides: std::collections::BTreeSet::new(),
90        });
91        Ok(())
92    }
93
94    pub async fn on_death(
95        &mut self,
96        id: Ulid,
97        status: ExitStatus,
98        _tx: &mpsc::Sender<GraphEvent>,
99    ) {
100        let Some(inc) = self.incarnated.remove(&id) else { return };
101        if let Some(p) = inc.pid {
102            self.by_pid.remove(&p.as_raw());
103        }
104        self.unregister_provider(&inc.card);
105        if let Some(parent) = inc.card.lineage {
106            if let Some(siblings) = self.children.get_mut(&parent) {
107                siblings.retain(|c| c != &id);
108            }
109        }
110        info!(label = %inc.card.label, ?status, "Ente disuelto");
111
112        match inc.card.supervision.clone() {
113            Supervision::Restart { initial, max: _ } => {
114                // Backoff exponencial: TODO real con timer del runtime.
115                tokio::time::sleep(initial).await;
116                let new_card = EntityCard { id: Ulid::new(), ..inc.card };
117                if let Err(e) = self.authorize_and_spawn(new_card, self.seed.id).await {
118                    warn!(?e, "restart falló");
119                }
120            }
121            Supervision::OneShot => {}
122            Supervision::Delegate => {
123                self.notify_lineage_of_death(&inc, &status);
124            }
125        }
126    }
127
128    /// Fire-and-forget: si el parent tiene conexión al bus, le forwardeamos
129    /// un Invoke con la muerte del hijo. Sin retry, sin backpressure.
130    fn notify_lineage_of_death(&mut self, inc: &Incarnated, status: &ExitStatus) {
131        let Some(parent) = inc.card.lineage else { return };
132        info!(
133            child = %inc.card.id, parent = %parent, label = %inc.card.label,
134            ?status,
135            "Supervision::Delegate — muerte notificada al lineage"
136        );
137        if let Some(out) = self.bus_connections.get(&parent).cloned() {
138            let blob = format!("{}:{:?}", inc.card.id, status);
139            let seq = self.alloc_invoke_seq();
140            let msg = BusMessage {
141                from: None,
142                seq,
143                payload: BusPayload::Request(BusRequest::Invoke {
144                    cap: Capability::Endpoint {
145                        interface: ente_card::InterfaceId([0xde; 16]),
146                        version: 1,
147                    },
148                    blob: blob.into_bytes(),
149                }),
150            };
151            let _ = out.try_send(msg);
152        }
153    }
154}