tawasuyu/arje

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 { card, pid });
88        Ok(())
89    }
90
91    pub async fn on_death(
92        &mut self,
93        id: Ulid,
94        status: ExitStatus,
95        _tx: &mpsc::Sender<GraphEvent>,
96    ) {
97        let Some(inc) = self.incarnated.remove(&id) else { return };
98        if let Some(p) = inc.pid {
99            self.by_pid.remove(&p.as_raw());
100        }
101        self.unregister_provider(&inc.card);
102        if let Some(parent) = inc.card.lineage {
103            if let Some(siblings) = self.children.get_mut(&parent) {
104                siblings.retain(|c| c != &id);
105            }
106        }
107        info!(label = %inc.card.label, ?status, "Ente disuelto");
108
109        match inc.card.supervision.clone() {
110            Supervision::Restart { initial, max: _ } => {
111                // Backoff exponencial: TODO real con timer del runtime.
112                tokio::time::sleep(initial).await;
113                let new_card = EntityCard { id: Ulid::new(), ..inc.card };
114                if let Err(e) = self.authorize_and_spawn(new_card, self.seed.id).await {
115                    warn!(?e, "restart falló");
116                }
117            }
118            Supervision::OneShot => {}
119            Supervision::Delegate => {
120                self.notify_lineage_of_death(&inc, &status);
121            }
122        }
123    }
124
125    /// Fire-and-forget: si el parent tiene conexión al bus, le forwardeamos
126    /// un Invoke con la muerte del hijo. Sin retry, sin backpressure.
127    fn notify_lineage_of_death(&mut self, inc: &Incarnated, status: &ExitStatus) {
128        let Some(parent) = inc.card.lineage else { return };
129        info!(
130            child = %inc.card.id, parent = %parent, label = %inc.card.label,
131            ?status,
132            "Supervision::Delegate — muerte notificada al lineage"
133        );
134        if let Some(out) = self.bus_connections.get(&parent).cloned() {
135            let blob = format!("{}:{:?}", inc.card.id, status);
136            let seq = self.alloc_invoke_seq();
137            let msg = BusMessage {
138                from: None,
139                seq,
140                payload: BusPayload::Request(BusRequest::Invoke {
141                    cap: Capability::Endpoint {
142                        interface: ente_card::InterfaceId([0xde; 16]),
143                        version: 1,
144                    },
145                    blob: blob.into_bytes(),
146                }),
147            };
148            let _ = out.try_send(msg);
149        }
150    }
151}