tawasuyu/arje

1//! ente-bus: bus de capacidades interno del fractal.
2//!
3//! Wire format: Unix SOCK_STREAM con framing length-prefijo (u32 BE) + payload
4//! postcard. Bidireccional pero por ahora request-response síncrono.
5//!
6//! Identidad: cada Ente hijo recibe `ENTE_BUS_SOCK` y `ENTE_ID` en su entorno.
7//! El cliente lee ambos vía `BusClient::from_env`.
8
9use ente_card::Capability;
10use serde::{Deserialize, Serialize};
11use std::path::Path;
12use std::str::FromStr;
13use tokio::io::{AsyncReadExt, AsyncWriteExt};
14use tokio::net::UnixStream;
15use ulid::Ulid;
16
17pub const ENV_BUS_SOCK: &str = "ENTE_BUS_SOCK";
18pub const ENV_ENTE_ID: &str = "ENTE_ID";
19pub const MAX_FRAME: usize = 1 << 20; // 1 MiB — protección contra OOM
20
21/// Credenciales del peer extraídas vía SO_PEERCRED al accept del bus.
22/// Imposibles de falsear desde el cliente — el kernel las inyecta.
23/// Definidas aquí (no en ente-zero) porque conceptualmente son atributo
24/// del protocolo del bus, no del init.
25#[derive(Debug, Clone, Copy)]
26pub struct PeerCreds {
27    pub pid: i32,
28    pub uid: u32,
29    pub gid: u32,
30}
31
32#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct BusMessage {
34    pub from: Option<Ulid>,
35    pub seq: u64,
36    pub payload: BusPayload,
37}
38
39#[derive(Debug, Clone, Serialize, Deserialize)]
40pub enum BusPayload {
41    Request(BusRequest),
42    Response(BusResponse),
43}
44
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub enum BusRequest {
47    /// Saludo. El Ente anuncia que está vivo y declara sus capacidades.
48    /// El Init usa esto para saber que el child arrancó correctamente,
49    /// independientemente de su exit code.
50    Announce { capabilities: Vec<Capability> },
51
52    /// Listar Entes vivos. Útil para debugging y para Entes-supervisor.
53    ListEntes,
54
55    /// Control de estado del fractal. Traducido desde D-Bus por compat-logind.
56    PowerOff { interactive: bool },
57    Reboot { interactive: bool },
58    Suspend { interactive: bool },
59    Hibernate { interactive: bool },
60
61    /// Invocación genérica de capacidad. `cap` debe estar provista por algún
62    /// Ente del grafo; el blob es el argumento opaco que el proveedor parsea.
63    Invoke { cap: Capability, blob: Vec<u8> },
64}
65
66#[derive(Debug, Clone, Serialize, Deserialize)]
67pub enum BusResponse {
68    Ok,
69    Error(String),
70    Entes(Vec<EnteInfo>),
71    Invoked { result: Vec<u8> },
72}
73
74#[derive(Debug, Clone, Serialize, Deserialize)]
75pub struct EnteInfo {
76    pub id: Ulid,
77    pub label: String,
78    pub provides: Vec<Capability>,
79    pub pid: Option<i32>,
80}
81
82pub async fn write_frame<W: AsyncWriteExt + Unpin>(w: &mut W, msg: &BusMessage) -> anyhow::Result<()> {
83    let bytes = postcard::to_stdvec(msg)?;
84    if bytes.len() > MAX_FRAME {
85        anyhow::bail!("frame too large: {} > {}", bytes.len(), MAX_FRAME);
86    }
87    w.write_u32(bytes.len() as u32).await?;
88    w.write_all(&bytes).await?;
89    Ok(())
90}
91
92pub async fn read_frame<R: AsyncReadExt + Unpin>(r: &mut R) -> anyhow::Result<BusMessage> {
93    let len = r.read_u32().await? as usize;
94    if len > MAX_FRAME {
95        anyhow::bail!("frame oversize: {len}");
96    }
97    let mut buf = vec![0u8; len];
98    r.read_exact(&mut buf).await?;
99    Ok(postcard::from_bytes(&buf)?)
100}
101
102pub struct BusClient {
103    stream: UnixStream,
104    seq: u64,
105    self_id: Option<Ulid>,
106}
107
108/// Trait que un Ente proveedor implementa para servir invokes que el bus le
109/// forwarda. Sync por simplicidad — un handler async se cubre con
110/// `tokio::task::block_in_place` o un canal hacia un task externo.
111pub trait InvokeHandler {
112    fn handle(&mut self, cap: ente_card::Capability, blob: Vec<u8>) -> BusResponse;
113}
114
115/// Conexión long-lived para Entes que proveen capacidades. A diferencia de
116/// `BusClient` (request-response y desconecta), `BusServer`:
117///   1. Anuncia su identidad al bus
118///   2. Mantiene la conexión abierta
119///   3. Atiende invokes forwardeados por el bus en bucle
120pub struct BusServer {
121    stream: UnixStream,
122    self_id: Ulid,
123}
124
125impl BusServer {
126    pub async fn from_env() -> anyhow::Result<Self> {
127        let path = std::env::var(ENV_BUS_SOCK)
128            .map_err(|_| anyhow::anyhow!("{} no definido", ENV_BUS_SOCK))?;
129        let id_s = std::env::var(ENV_ENTE_ID)
130            .map_err(|_| anyhow::anyhow!("{} no definido", ENV_ENTE_ID))?;
131        let self_id = Ulid::from_str(&id_s)
132            .map_err(|_| anyhow::anyhow!("{} no es un Ulid válido: {id_s}", ENV_ENTE_ID))?;
133        let stream = UnixStream::connect(&path).await?;
134        Ok(Self { stream, self_id })
135    }
136
137    pub async fn announce(&mut self, capabilities: Vec<ente_card::Capability>) -> anyhow::Result<()> {
138        let req = BusMessage {
139            from: Some(self.self_id),
140            seq: 1,
141            payload: BusPayload::Request(BusRequest::Announce { capabilities }),
142        };
143        write_frame(&mut self.stream, &req).await?;
144        let resp = read_frame(&mut self.stream).await?;
145        match resp.payload {
146            BusPayload::Response(BusResponse::Ok) => Ok(()),
147            BusPayload::Response(other) => anyhow::bail!("Announce rechazado: {other:?}"),
148            BusPayload::Request(_) => anyhow::bail!("expected Response, got Request"),
149        }
150    }
151
152    /// Bucle principal del proveedor. Atiende invokes hasta que la conexión
153    /// se cierra (el bus muere o el Ente recibe SIGTERM y nosotros dropeamos).
154    pub async fn serve<H: InvokeHandler>(mut self, mut handler: H) -> anyhow::Result<()> {
155        loop {
156            let msg = read_frame(&mut self.stream).await?;
157            let resp = match msg.payload {
158                BusPayload::Request(BusRequest::Invoke { cap, blob }) => {
159                    handler.handle(cap, blob)
160                }
161                BusPayload::Request(other) => {
162                    BusResponse::Error(format!("BusServer no maneja {other:?}"))
163                }
164                BusPayload::Response(_) => continue,
165            };
166            let out = BusMessage {
167                from: Some(self.self_id),
168                seq: msg.seq,
169                payload: BusPayload::Response(resp),
170            };
171            write_frame(&mut self.stream, &out).await?;
172        }
173    }
174}
175
176impl BusClient {
177    pub async fn connect(path: impl AsRef<Path>) -> anyhow::Result<Self> {
178        let stream = UnixStream::connect(path).await?;
179        let self_id = std::env::var(ENV_ENTE_ID)
180            .ok()
181            .and_then(|s| Ulid::from_str(&s).ok());
182        Ok(Self { stream, seq: 0, self_id })
183    }
184
185    pub async fn from_env() -> anyhow::Result<Self> {
186        let path = std::env::var(ENV_BUS_SOCK)
187            .map_err(|_| anyhow::anyhow!("{} no definido", ENV_BUS_SOCK))?;
188        Self::connect(&path).await
189    }
190
191    pub async fn call(&mut self, req: BusRequest) -> anyhow::Result<BusResponse> {
192        self.seq = self.seq.wrapping_add(1);
193        let msg = BusMessage {
194            from: self.self_id,
195            seq: self.seq,
196            payload: BusPayload::Request(req),
197        };
198        write_frame(&mut self.stream, &msg).await?;
199        let resp = read_frame(&mut self.stream).await?;
200        match resp.payload {
201            BusPayload::Response(r) => Ok(r),
202            BusPayload::Request(_) => anyhow::bail!("expected response, got request"),
203        }
204    }
205}