sergio/takana

1//! Bus de agente: `/run/agent.sock`. Ver `docs/07-agent-bus.md`.
2//!
3//! Modelo de threading (síncrono, una conexión = N threads pequeños):
4//!
5//! ```text
6//!   [accept loop] ── spawn ──▶ conn{
7//!        [reader thread] ── dispatch ──┐
8//!        [bus forwarder]  ─────────────┼──▶ tx ──▶ [writer thread] ──▶ socket
9//!        [compile worker] (puntual) ───┘
10//!   }
11//! ```
12//!
13//! No usamos `tokio` para no arrastrar runtime asíncrono: la cadencia esperada del bus es
14//! humana + agente (decenas de mensajes/seg en el peor caso). Cada conexión tiene su propio
15//! canal `Sender<Event>`; la `EventBus` global re-emite Modified/Crashed a esos canales.
16
17use std::io::{BufRead, BufReader, Write};
18use std::os::fd::AsFd;
19use std::os::unix::net::{UnixListener, UnixStream};
20use std::path::{Path, PathBuf};
21use std::sync::mpsc::{self, Sender};
22use std::sync::{Arc, Mutex};
23use std::thread;
24
25use nix::sys::socket::{getsockopt, sockopt::PeerCredentials};
26
27use crate::events::EventBus;
28use hammer_core::proto::{Cap, Command, Event, Peer, RecipeInline, PROTOCOL_VERSION};
29
30/// Política de capacidades: dado el `Peer` (ya autenticado por SO_PEERCRED), devuelve qué
31/// caps tiene esa conexión. Por defecto: misma UID que el daemon ⇒ todo menos `inject-real`;
32/// otros UIDs ⇒ sólo `query`. La policy más expresiva (lectura de un toml en `/etc/hammer`)
33/// es trabajo posterior; la firma del callable basta para encajarla sin tocar este módulo.
34pub type CapsPolicy = Arc<dyn Fn(&Peer) -> Vec<Cap> + Send + Sync>;
35
36pub fn default_policy() -> CapsPolicy {
37    let my_uid = unsafe { libc::getuid() };
38    Arc::new(move |peer: &Peer| {
39        if peer.uid == my_uid {
40            vec![Cap::Query, Cap::Compile, Cap::Inject, Cap::Init]
41        } else {
42            vec![Cap::Query]
43        }
44    })
45}
46
47/// Contexto que necesita el dispatcher para resolver comandos. Compartido por todas las
48/// conexiones (clones baratos: paths + Arc).
49#[derive(Clone)]
50pub struct BusContext {
51    pub store_root: PathBuf,
52    pub init_control: PathBuf,
53    pub events: EventBus,
54}
55
56#[derive(Debug, thiserror::Error)]
57pub enum Error {
58    #[error("io: {0}")]
59    Io(#[from] std::io::Error),
60    #[error("bind {0}: {1}")]
61    Bind(PathBuf, std::io::Error),
62    #[error("peer creds: {0}")]
63    Peer(String),
64}
65
66pub type Result<T> = std::result::Result<T, Error>;
67
68/// Arranca el listener bloqueante. Pensado para correr en su propio thread desde `main`.
69pub fn serve_agent_bus(
70    sock_path: &Path,
71    policy: CapsPolicy,
72    ctx: BusContext,
73) -> Result<()> {
74    // Limpia un socket olvidado de una ejecución previa. Si el path está ocupado por algo
75    // que no es un socket, dejamos que `bind` falle — no queremos rm-rf por accidente.
76    let _ = std::fs::remove_file(sock_path);
77    let listener = UnixListener::bind(sock_path)
78        .map_err(|e| Error::Bind(sock_path.to_path_buf(), e))?;
79    tracing::info!(sock = %sock_path.display(), "bus: escuchando");
80
81    for stream in listener.incoming() {
82        match stream {
83            Ok(s) => {
84                let policy = policy.clone();
85                let ctx = ctx.clone();
86                thread::Builder::new()
87                    .name("bus-conn".into())
88                    .spawn(move || {
89                        if let Err(e) = handle_connection(s, policy, ctx) {
90                            tracing::warn!(error = %e, "bus: conexión terminó con error");
91                        }
92                    })
93                    .ok();
94            }
95            Err(e) => tracing::warn!(error = %e, "bus: accept falló"),
96        }
97    }
98    Ok(())
99}
100
101fn read_peer(stream: &UnixStream) -> Result<Peer> {
102    let cred = getsockopt(&stream.as_fd(), PeerCredentials)
103        .map_err(|e| Error::Peer(e.to_string()))?;
104    Ok(Peer {
105        uid: cred.uid(),
106        gid: cred.gid(),
107        pid: cred.pid(),
108    })
109}
110
111fn handle_connection(stream: UnixStream, policy: CapsPolicy, ctx: BusContext) -> Result<()> {
112    let peer = read_peer(&stream)?;
113    let caps = policy(&peer);
114    tracing::info!(?peer, ?caps, "bus: nueva conexión");
115
116    // Canal de eventos hacia el writer. Todos los productores (dispatcher, bus forwarder,
117    // workers de COMPILE) clonan este Sender.
118    let (tx, rx) = mpsc::channel::<Event>();
119
120    // Writer thread: serializa todo lo que recibe del canal y lo escribe al socket.
121    let stream_writer = stream.try_clone()?;
122    let _writer_handle = thread::Builder::new()
123        .name("bus-conn-writer".into())
124        .spawn(move || writer_loop(stream_writer, rx))
125        .ok();
126
127    // Bus forwarder thread: cualquier evento global (Modified/Crashed) se re-emite a este
128    // peer. Se cierra solo cuando el global drop el último Sender (en la práctica, nunca)
129    // o cuando nuestro tx local cae (peer cerró).
130    let bus_rx = ctx.events.subscribe();
131    let bus_tx = tx.clone();
132    thread::Builder::new()
133        .name("bus-conn-fwd".into())
134        .spawn(move || {
135            for ev in bus_rx {
136                if bus_tx.send(ev).is_err() {
137                    break;
138                }
139            }
140        })
141        .ok();
142
143    // Handshake. La primera línea debe ser Hello.
144    let mut reader = BufReader::new(stream);
145    let mut first = String::new();
146    reader.read_line(&mut first)?;
147    match serde_json::from_str::<Command>(first.trim()) {
148        Ok(Command::Hello { ver, client }) => {
149            tracing::info!(ver, client = %client, "bus: hello");
150            let _ = tx.send(Event::Welcome {
151                ver: PROTOCOL_VERSION,
152                caps: caps.clone(),
153                peer: peer.clone(),
154            });
155        }
156        Ok(other) => {
157            let _ = tx.send(Event::Error {
158                code: "no_hello".into(),
159                msg: format!("se esperaba 'hello', llegó '{}'", t_of(&other)),
160            });
161            return Ok(());
162        }
163        Err(e) => {
164            let _ = tx.send(Event::Error {
165                code: "bad_handshake".into(),
166                msg: format!("hello inválido: {e}"),
167            });
168            return Ok(());
169        }
170    }
171
172    // Read loop. Cada línea es un Command. La conexión se cierra cuando se cierra EOF, o
173    // cuando un error de IO lo fuerza.
174    let caps_set: std::collections::BTreeSet<Cap> = caps.iter().copied().collect();
175    let caps_set = Arc::new(caps_set);
176    for line in reader.lines() {
177        let line = match line {
178            Ok(l) => l,
179            Err(_) => break,
180        };
181        if line.trim().is_empty() {
182            continue;
183        }
184        let cmd: Command = match serde_json::from_str(&line) {
185            Ok(c) => c,
186            Err(e) => {
187                let _ = tx.send(Event::Error {
188                    code: "bad_command".into(),
189                    msg: format!("JSON inválido: {e}"),
190                });
191                continue;
192            }
193        };
194        dispatch(cmd, &tx, &caps_set, &ctx);
195    }
196    Ok(())
197}
198
199fn writer_loop(mut stream: UnixStream, rx: mpsc::Receiver<Event>) {
200    for ev in rx {
201        let line = match serde_json::to_string(&ev) {
202            Ok(s) => s,
203            Err(e) => {
204                tracing::error!(error = %e, "bus: serializando event");
205                continue;
206            }
207        };
208        if stream.write_all(line.as_bytes()).is_err() {
209            break;
210        }
211        if stream.write_all(b"\n").is_err() {
212            break;
213        }
214        // El bus es de baja cadencia; flush por mensaje no es relevante en costes y mejora
215        // mucho la latencia percibida del cliente.
216        let _ = stream.flush();
217    }
218}
219
220fn dispatch(
221    cmd: Command,
222    tx: &Sender<Event>,
223    caps: &Arc<std::collections::BTreeSet<Cap>>,
224    ctx: &BusContext,
225) {
226    if let Some(needed) = Cap::required_for(&cmd) {
227        if !caps.contains(&needed) {
228            let _ = tx.send(Event::Error {
229                code: "no_cap".into(),
230                msg: format!("comando requiere capacidad '{needed:?}', no concedida"),
231            });
232            return;
233        }
234    }
235    match cmd {
236        Command::Hello { .. } => {
237            // Doble hello: lo ignoramos con un Error suave.
238            let _ = tx.send(Event::Error {
239                code: "duplicate_hello".into(),
240                msg: "ya hicimos hello en esta conexión".into(),
241            });
242        }
243        Command::Compile { recipe } => {
244            // Build pesado: a un worker. El read-loop sigue libre para más comandos.
245            let tx = tx.clone();
246            let store_root = ctx.store_root.clone();
247            thread::Builder::new()
248                .name("bus-compile".into())
249                .spawn(move || run_compile(recipe, store_root, tx))
250                .ok();
251        }
252        Command::Inject { artifact, target, overlay: _ } => {
253            let store = match hammer_core::Store::open(&ctx.store_root) {
254                Ok(s) => s,
255                Err(e) => {
256                    let _ = tx.send(Event::Error {
257                        code: "store".into(),
258                        msg: e.to_string(),
259                    });
260                    return;
261                }
262            };
263            let artifact_dir = match store.find_by_hash(&artifact) {
264                Ok(p) => p,
265                Err(e) => {
266                    let _ = tx.send(Event::Error {
267                        code: "not_found".into(),
268                        msg: e.to_string(),
269                    });
270                    return;
271                }
272            };
273            let report = match hammer_build::run_hydrate(
274                &artifact_dir,
275                Path::new(&target),
276                hammer_core::LinkMode::Static,
277            ) {
278                Ok(r) => r,
279                Err(e) => {
280                    let _ = tx.send(Event::Error {
281                        code: "hydrate_failed".into(),
282                        msg: e.to_string(),
283                    });
284                    return;
285                }
286            };
287            let _ = tx.send(Event::Injected {
288                artifact,
289                target,
290                files: report.files.len(),
291            });
292        }
293        Command::Query { what, path, name } => {
294            let value = match what.as_str() {
295                "file" => {
296                    let Some(p) = path else {
297                        let _ = tx.send(Event::Error {
298                            code: "bad_query".into(),
299                            msg: "query 'file' requiere 'path'".into(),
300                        });
301                        return;
302                    };
303                    query_file(&p)
304                }
305                "artifact" => {
306                    let Some(h) = name.as_deref().or(path.as_deref()) else {
307                        let _ = tx.send(Event::Error {
308                            code: "bad_query".into(),
309                            msg: "query 'artifact' requiere 'name' (el hash)".into(),
310                        });
311                        return;
312                    };
313                    query_artifact(&ctx.store_root, h)
314                }
315                other => {
316                    let _ = tx.send(Event::Error {
317                        code: "unknown_query".into(),
318                        msg: format!("'what'='{other}' no soportado"),
319                    });
320                    return;
321                }
322            };
323            let _ = tx.send(Event::QueryResult { what, value });
324        }
325        Command::Init { cmd } => {
326            match crate::control::send_to_fifo(&ctx.init_control, &cmd) {
327                Ok(()) => {
328                    let _ = tx.send(Event::InitAck { cmd });
329                }
330                Err(e) => {
331                    let _ = tx.send(Event::Error {
332                        code: "init_send_failed".into(),
333                        msg: e.to_string(),
334                    });
335                }
336            }
337        }
338    }
339}
340
341fn run_compile(recipe: RecipeInline, store_root: PathBuf, tx: Sender<Event>) {
342    let name = recipe.name.clone();
343    let mutation = hammer_core::swm::Mutation::SourcePatch {
344        repo: recipe.repo,
345        commit: recipe.commit,
346        patch: recipe.patch,
347        patch_url: None,
348        build: hammer_core::swm::SwmBuild {
349            compiler: recipe.compiler,
350            target: recipe.target,
351            link: recipe.link,
352            flags: recipe.flags,
353        },
354        target_bin: format!("/usr/bin/{name}"),
355        expected_hash: None,
356    };
357    let store = match hammer_core::Store::open(&store_root) {
358        Ok(s) => s,
359        Err(e) => {
360            let _ = tx.send(Event::BuildFailed {
361                recipe: name,
362                reason: format!("store: {e}"),
363                log_tail: None,
364            });
365            return;
366        }
367    };
368    let cfg = hammer_build::BuildConfig::from_env_or_defaults(store.root());
369    match hammer_build::build_source_patch(&mutation, &cfg, &store, None) {
370        Ok(h) => {
371            let _ = tx.send(Event::BuildReady {
372                recipe: name,
373                artifact: h.as_str().to_string(),
374            });
375        }
376        Err(e) => {
377            let _ = tx.send(Event::BuildFailed {
378                recipe: name,
379                reason: e.to_string(),
380                log_tail: None,
381            });
382        }
383    }
384}
385
386fn query_file(path: &str) -> serde_json::Value {
387    let p = Path::new(path);
388    let meta = match std::fs::symlink_metadata(p) {
389        Ok(m) => m,
390        Err(e) => return serde_json::json!({"exists": false, "error": e.to_string()}),
391    };
392    use std::os::unix::fs::MetadataExt;
393    serde_json::json!({
394        "exists": true,
395        "size": meta.size(),
396        "mode": meta.mode(),
397        "uid": meta.uid(),
398        "gid": meta.gid(),
399        "is_dir": meta.is_dir(),
400        "is_symlink": meta.file_type().is_symlink(),
401    })
402}
403
404fn query_artifact(store_root: &Path, hash: &str) -> serde_json::Value {
405    let store = match hammer_core::Store::open(store_root) {
406        Ok(s) => s,
407        Err(e) => return serde_json::json!({"error": e.to_string()}),
408    };
409    match store.find_by_hash(hash) {
410        Ok(p) => serde_json::json!({"exists": true, "path": p.display().to_string()}),
411        Err(_) => serde_json::json!({"exists": false}),
412    }
413}
414
415fn t_of(cmd: &Command) -> &'static str {
416    match cmd {
417        Command::Hello { .. } => "hello",
418        Command::Compile { .. } => "compile",
419        Command::Inject { .. } => "inject",
420        Command::Query { .. } => "query",
421        Command::Init { .. } => "init",
422    }
423}
424
425// Re-export para que el caller no tenga que importar Mutex/Arc directamente.
426#[allow(dead_code)]
427pub(crate) type SharedCaps = Arc<Mutex<std::collections::BTreeSet<Cap>>>;
428
429#[cfg(test)]
430mod tests {
431    use super::*;
432
433    #[test]
434    fn default_policy_grants_query_to_others() {
435        let p = default_policy();
436        // Un peer con UID distinto al actual debe quedar como sólo-query.
437        let foreign = Peer {
438            uid: u32::MAX,
439            gid: 0,
440            pid: 1,
441        };
442        assert_eq!(p(&foreign), vec![Cap::Query]);
443    }
444
445    #[test]
446    fn t_of_matches_all_commands() {
447        // No es exhaustivo de Cargo, pero verifica que no nos olvidemos del nombre del tag.
448        assert_eq!(t_of(&Command::Hello { ver: 1, client: "".into() }), "hello");
449        assert_eq!(
450            t_of(&Command::Compile {
451                recipe: RecipeInline {
452                    name: "x".into(),
453                    repo: "g".into(),
454                    commit: "c".into(),
455                    patch: None,
456                    compiler: "zig-cc".into(),
457                    target: "t".into(),
458                    link: "static".into(),
459                    flags: vec![],
460                },
461            }),
462            "compile"
463        );
464    }
465
466    #[test]
467    fn query_file_reports_missing() {
468        let v = query_file("/no/existe/seguramente/ahora");
469        assert_eq!(v["exists"], serde_json::json!(false));
470    }
471}