⎇ c22d2480b93b7802be8e8ea5a0075b8175c4fedc
crates/modules/shipote/shipote-core/src/pipeline.rs
History
Who touched what
raw
1//! Pipeline runtime: encadena nodos con pipes y opcionalmente intercepta
2//! cada flow para discernir su contenido.
3//!
4//! Cada nodo se encarna via [`ente_incarnate::Incarnator`] — eso significa
5//! que **cada comando puede tener su propio SomaSpec** (namespaces, cgroup,
6//! rlimits) heredado del workspace. La conexión stdin↔stdout se hace con
7//! `pipe2(2)` + `ChildStdio` declarativo: el callback de clone(2) hace los
8//! `dup2` pre-execve sin romper la regla async-signal-safe.
9
10use crate::CoreError;
11use brahman_card::Payload;
12use ente_incarnate::{ChildStdio, Incarnator};
13use nix::fcntl::OFlag;
14use nix::unistd::pipe2;
15use shipote_card::{FlowEdge, PipelineSpec};
16use shipote_discern::{DiscernPipeline, Discernment, Hint};
17use std::os::fd::{AsRawFd, IntoRawFd, RawFd};
18use std::sync::Arc;
19use tokio::io::unix::AsyncFd;
20use tokio::io::Interest;
21use tracing::{debug, info, warn};
22use ulid::Ulid;
23
24/// Resultado de lanzar un pipeline.
25#[derive(Debug, Clone)]
26pub struct PipelineLaunch {
27 pub pipeline: Ulid,
28 pub command_pids: Vec<(String, i32)>,
29 /// Discernments por edge, en el mismo orden que `spec.edges`.
30 pub edge_discernments: Vec<EdgeDiscernment>,
31}
32
33#[derive(Debug, Clone)]
34pub struct EdgeDiscernment {
35 pub from_label: String,
36 pub from_output: String,
37 pub to_label: String,
38 pub to_input: String,
39 pub discernment: Option<Discernment>,
40}
41
42/// Lanza un pipeline conectando nodos por stdin/stdout. Cada nodo se
43/// encarna via `Incarnator` (con o sin namespacing según su SomaSpec).
44///
45/// v1: pipeline lineal (un edge entrante por nodo). Múltiples edges
46/// entrantes generan warning y sólo el primero se honra.
47pub async fn run_pipeline(
48 spec: &PipelineSpec,
49 workspace_label: &str,
50 tap: bool,
51 discerner: Arc<DiscernPipeline>,
52 incarnator: Arc<Incarnator>,
53) -> Result<PipelineLaunch, CoreError> {
54 spec.validate()?;
55 let n = spec.nodes.len();
56 info!(
57 nodes = n,
58 edges = spec.edges.len(),
59 tap,
60 "launching pipeline (incarnated)"
61 );
62
63 // Predecessor: para cada nodo, su edge entrante (si tiene).
64 let mut predecessor: Vec<Option<&FlowEdge>> = vec![None; n];
65 for e in &spec.edges {
66 if predecessor[e.to].is_some() {
67 warn!(node = e.to, "v1 pipeline: nodo con múltiples predecessors — sólo se honra el primero");
68 continue;
69 }
70 predecessor[e.to] = Some(e);
71 }
72
73 let mut pids = Vec::with_capacity(n);
74 let mut taps: Vec<TapHandle> = Vec::new();
75 // Para cada nodo i que produce, guardamos el FD de read del pipe
76 // del productor → al armar el consumidor lo consume.
77 // Pero como puede haber tap intermedio, llevamos un esquema:
78 // - Sin tap: read FD del pipe productor → stdin del consumidor.
79 // - Con tap: read FD del pipe productor → tokio proxy → write FD
80 // del pipe consumidor → stdin del consumidor.
81 // Para simplicidad lineal, `pending_stdin_for_next` guarda el FD que
82 // el siguiente consumidor debe usar como stdin.
83 let mut pending_stdin_for_next: Option<RawFd> = None;
84
85 for (i, node) in spec.nodes.iter().enumerate() {
86 // Validar payload ejecutable.
87 match &node.payload {
88 Payload::Native { .. } | Payload::Legacy { .. } => {}
89 _ => {
90 return Err(CoreError::Incarnate(
91 ente_incarnate::IncarnateError::NonExecutablePayload,
92 ))
93 }
94 }
95
96 // Compilamos a Card.
97 let card = node.to_card(i, workspace_label)?;
98
99 // ¿Soy productor? Necesito stdout_fd hacia un pipe nuevo.
100 let i_is_producer = spec.edges.iter().any(|e| e.from == i);
101 let stdin_fd: Option<RawFd> = pending_stdin_for_next.take();
102 let mut stdout_fd: Option<RawFd> = None;
103 let mut next_pending: Option<RawFd> = None;
104
105 // FDs que el PADRE debe cerrar tras spawn (son nuestra copia del
106 // extremo que pasamos al hijo).
107 let mut parent_closes: Vec<RawFd> = Vec::new();
108
109 if i_is_producer {
110 let (r, w) = pipe2(OFlag::O_CLOEXEC).map_err(|e| {
111 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
112 })?;
113 let r_raw = r.into_raw_fd();
114 let w_raw = w.into_raw_fd();
115 stdout_fd = Some(w_raw);
116 parent_closes.push(w_raw);
117
118 if tap {
119 // Necesitamos un segundo pipe entre tap y consumidor.
120 let (r2, w2) = pipe2(OFlag::O_CLOEXEC).map_err(|e| {
121 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
122 })?;
123 let r2_raw = r2.into_raw_fd();
124 let w2_raw = w2.into_raw_fd();
125 next_pending = Some(r2_raw);
126 // El tap lee de r_raw y escribe a w2_raw.
127 let edge = predecessor
128 .iter()
129 .find_map(|p| *p)
130 .and_then(|e| if e.from == i { Some(e) } else { None })
131 // Edge donde i es from:
132 .or_else(|| spec.edges.iter().find(|e| e.from == i));
133 let from_label = node.label.clone();
134 let to_label = edge
135 .map(|e| spec.nodes[e.to].label.clone())
136 .unwrap_or_default();
137 let from_output = edge.map(|e| e.from_output.clone()).unwrap_or_default();
138 let to_input = edge.map(|e| e.to_input.clone()).unwrap_or_default();
139 let sample_bytes = spec.discern.sample_bytes;
140 let disc = discerner.clone();
141 let h = spawn_tap(
142 r_raw, w2_raw, sample_bytes, disc, from_label, from_output, to_label, to_input,
143 );
144 taps.push(h);
145 // r_raw y w2_raw pasaron a manos del tokio task. No los
146 // cerramos en el padre.
147 } else {
148 // Sin tap, el read del productor va directo al stdin del
149 // siguiente consumidor.
150 next_pending = Some(r_raw);
151 }
152 }
153
154 let stdio = ChildStdio {
155 stdin_fd,
156 stdout_fd,
157 stderr_fd: None,
158 };
159
160 // Incarnator absorbe los fds de `stdio` — no los cerramos acá.
161 // `parent_closes` queda obsoleto.
162 let _ = parent_closes;
163 let outcome = incarnator
164 .incarnate_with(&card, stdio)
165 .map_err(CoreError::Incarnate)?;
166 let pid = outcome.pid;
167 pids.push((node.label.clone(), pid.as_raw()));
168 debug!(label = %node.label, pid = pid.as_raw(), "node incarnated");
169
170 pending_stdin_for_next = next_pending;
171 }
172
173 let pipeline_id = Ulid::new();
174
175 let mut edge_discernments = Vec::with_capacity(taps.len());
176 for t in taps {
177 match t.handle.await {
178 Ok(d) => edge_discernments.push(d),
179 Err(e) => warn!(?e, "tap handle joined with error"),
180 }
181 }
182
183 Ok(PipelineLaunch {
184 pipeline: pipeline_id,
185 command_pids: pids,
186 edge_discernments,
187 })
188}
189
190struct TapHandle {
191 handle: tokio::task::JoinHandle<EdgeDiscernment>,
192}
193
194#[allow(clippy::too_many_arguments)]
195fn spawn_tap(
196 producer_r_fd: RawFd,
197 consumer_w_fd: RawFd,
198 sample_bytes: usize,
199 discerner: Arc<DiscernPipeline>,
200 from_label: String,
201 from_output: String,
202 to_label: String,
203 to_input: String,
204) -> TapHandle {
205 // Marcar non-blocking ANTES de envolverlos en AsyncFd. Sino tokio
206 // bloquea el reactor en operaciones lentas.
207 set_nonblocking(producer_r_fd);
208 set_nonblocking(consumer_w_fd);
209
210 let handle = tokio::spawn(async move {
211 // SAFETY: el caller transfiere ownership de los fds al task.
212 let r_std = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(producer_r_fd) };
213 let w_std = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(consumer_w_fd) };
214 let r = AsyncFd::with_interest(r_std, Interest::READABLE).expect("AsyncFd r");
215 let w = AsyncFd::with_interest(w_std, Interest::WRITABLE).expect("AsyncFd w");
216
217 let mut sample: Vec<u8> = Vec::with_capacity(sample_bytes);
218 let mut buf = [0u8; 4096];
219 let mut total: u64 = 0;
220
221 // Fase 1: sampling + pump.
222 let mut eof = false;
223 while !eof && sample.len() < sample_bytes {
224 let n = match async_read(&r, &mut buf).await {
225 Ok(0) => { eof = true; 0 }
226 Ok(n) => n,
227 Err(e) => { warn!(?e, "tap producer read failed"); break; }
228 };
229 if n == 0 { break; }
230 let take = n.min(sample_bytes - sample.len());
231 sample.extend_from_slice(&buf[..take]);
232 if let Err(e) = async_write_all(&w, &buf[..n]).await {
233 warn!(?e, "tap consumer write failed");
234 break;
235 }
236 total += n as u64;
237 }
238 let d = discerner.discern(&sample, &Hint { path: None, size_total: None });
239
240 // Fase 2: pump-only hasta EOF.
241 while !eof {
242 let n = match async_read(&r, &mut buf).await {
243 Ok(0) => { eof = true; 0 }
244 Ok(n) => n,
245 Err(_) => break,
246 };
247 if n == 0 { break; }
248 if async_write_all(&w, &buf[..n]).await.is_err() { break; }
249 total += n as u64;
250 }
251 debug!(bytes = total, "tap finished");
252 EdgeDiscernment {
253 from_label,
254 from_output,
255 to_label,
256 to_input,
257 discernment: d,
258 }
259 });
260 TapHandle { handle }
261}
262
263async fn async_read(
264 afd: &AsyncFd<std::os::fd::OwnedFd>,
265 buf: &mut [u8],
266) -> std::io::Result<usize> {
267 loop {
268 let mut guard = afd.readable().await?;
269 let fd = afd.as_raw_fd();
270 // SAFETY: lectura sobre fd válido propiedad del AsyncFd.
271 let r = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
272 if r >= 0 {
273 return Ok(r as usize);
274 }
275 let err = std::io::Error::last_os_error();
276 if err.kind() == std::io::ErrorKind::WouldBlock {
277 guard.clear_ready();
278 continue;
279 }
280 return Err(err);
281 }
282}
283
284async fn async_write_all(
285 afd: &AsyncFd<std::os::fd::OwnedFd>,
286 mut buf: &[u8],
287) -> std::io::Result<()> {
288 while !buf.is_empty() {
289 let mut guard = afd.writable().await?;
290 let fd = afd.as_raw_fd();
291 // SAFETY: escritura sobre fd válido propiedad del AsyncFd.
292 let r = unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) };
293 if r > 0 {
294 buf = &buf[r as usize..];
295 continue;
296 }
297 if r == 0 {
298 return Err(std::io::Error::new(
299 std::io::ErrorKind::WriteZero,
300 "write 0",
301 ));
302 }
303 let err = std::io::Error::last_os_error();
304 if err.kind() == std::io::ErrorKind::WouldBlock {
305 guard.clear_ready();
306 continue;
307 }
308 return Err(err);
309 }
310 Ok(())
311}
312
313fn set_nonblocking(fd: RawFd) {
314 // SAFETY: fcntl con F_SETFL es seguro para fds válidos.
315 unsafe {
316 let flags = libc::fcntl(fd, libc::F_GETFL, 0);
317 if flags >= 0 {
318 libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
319 }
320 }
321}
322
323// Extension trait para abstraer la API de OwnedFd entre versiones (compat).
324trait OwnedFdFromRawCompat: Sized {
325 unsafe fn from_raw_fd_compat(fd: RawFd) -> Self;
326}
327
328impl OwnedFdFromRawCompat for std::os::fd::OwnedFd {
329 unsafe fn from_raw_fd_compat(fd: RawFd) -> Self {
330 use std::os::fd::FromRawFd;
331 // SAFETY: el caller transfiere ownership de `fd` a la `OwnedFd`.
332 unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }
333 }
334}
335
336// Re-export para que el unused warning del AsRawFd se calle si no se usa.
337#[allow(dead_code)]
338fn _keep_raw(_: &dyn AsRawFd) {}
339
340#[cfg(test)]
341mod tests {
342 use super::*;
343 use brahman_card::Payload;
344 use ente_incarnate::IncarnatorConfig;
345 use shipote_card::{CommandRef, DiscernPolicy, FlowEdge, PipelineSpec, WorkspaceId};
346
347 fn cmd(label: &str, exec: &str, argv: &[&str]) -> CommandRef {
348 CommandRef {
349 label: label.into(),
350 payload: Payload::Native {
351 exec: exec.into(),
352 argv: argv.iter().map(|s| s.to_string()).collect(),
353 envp: vec![],
354 },
355 soma: Default::default(),
356 flows: Default::default(),
357 supervision: brahman_card::Supervision::OneShot,
358 }
359 }
360
361 #[tokio::test]
362 async fn pipeline_isolated_echo_to_cat_runs() {
363 let spec = PipelineSpec {
364 label: "echo-cat".into(),
365 workspace: WorkspaceId::new(),
366 nodes: vec![
367 cmd("p1", "/bin/echo", &["hola pipeline aislado"]),
368 cmd("p2", "/bin/cat", &[]),
369 ],
370 edges: vec![FlowEdge {
371 from: 0,
372 from_output: "stdout".into(),
373 to: 1,
374 to_input: "stdin".into(),
375 }],
376 discern: DiscernPolicy::default(),
377 };
378 let disc = Arc::new(DiscernPipeline::default_pipeline());
379 let inc = Arc::new(Incarnator::new(IncarnatorConfig::default()));
380 let launch = run_pipeline(&spec, "ws", false, disc, inc).await.unwrap();
381 assert_eq!(launch.command_pids.len(), 2);
382 // Cosecha.
383 for (_, pid) in &launch.command_pids {
384 let _ = nix::sys::wait::waitpid(nix::unistd::Pid::from_raw(*pid), None);
385 }
386 }
387
388 #[tokio::test]
389 async fn pipeline_isolated_with_tap_captures_discernment() {
390 let spec = PipelineSpec {
391 label: "json-cat".into(),
392 workspace: WorkspaceId::new(),
393 nodes: vec![
394 cmd("p1", "/bin/echo", &["{\"hello\": 1}"]),
395 cmd("p2", "/bin/cat", &[]),
396 ],
397 edges: vec![FlowEdge {
398 from: 0,
399 from_output: "stdout".into(),
400 to: 1,
401 to_input: "stdin".into(),
402 }],
403 discern: DiscernPolicy {
404 sample_bytes: 4096,
405 enrich_producer: true,
406 },
407 };
408 let disc = Arc::new(DiscernPipeline::default_pipeline());
409 let inc = Arc::new(Incarnator::new(IncarnatorConfig::default()));
410 let launch = run_pipeline(&spec, "ws", true, disc, inc).await.unwrap();
411 assert_eq!(launch.edge_discernments.len(), 1);
412 let d = &launch.edge_discernments[0];
413 let dis = d.discernment.as_ref().expect("discernment present");
414 assert_eq!(dis.mime.as_deref(), Some("application/json"));
415 // Cosecha.
416 for (_, pid) in &launch.command_pids {
417 let _ = nix::sys::wait::waitpid(nix::unistd::Pid::from_raw(*pid), None);
418 }
419 }
420}