⎇ d8727a3038c45b4fc69a37fc988e38408ddb5f6c
crates/modules/shipote/shipote-core/src/pipeline.rs
Historia
Quién tocó qué
crudo
feat(shipote): CPU% + pipeline live-tail + replay por bytes (fase J)
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::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)]
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 /// Path del Unix socket donde otros módulos pueden suscribirse al
41 /// stream replicado por este edge. `None` cuando tap=false (no hay
42 /// data plane porque no hay sampling).
43 pub flow_socket: Option<std::path::PathBuf>,
44}
45
46/// Lanza un pipeline conectando nodos por stdin/stdout. Cada nodo se
47/// encarna via `Incarnator` (con o sin namespacing según su SomaSpec).
48///
49/// Soporta:
50/// - Pipeline lineal (1 producer → 1 consumer).
51/// - **Fan-out** (1 producer → N consumers): shipote interpone un
52/// splitter que duplica bytes a cada destino. Cuando `tap=true`, el
53/// splitter además samplea para discernir.
54/// - Múltiples predecessors por nodo NO se soporta aún (fan-in): sólo se
55/// honra el primer edge entrante.
56pub async fn run_pipeline(
57 spec: &PipelineSpec,
58 workspace_label: &str,
59 tap: bool,
60 discerner: Arc<DiscernPipeline>,
61 incarnator: Arc<Incarnator>,
62 manager: Option<Arc<crate::WorkspaceManager>>,
63) -> Result<PipelineLaunch, CoreError> {
64 spec.validate()?;
65 let n = spec.nodes.len();
66 info!(
67 nodes = n,
68 edges = spec.edges.len(),
69 tap,
70 "launching pipeline (incarnated)"
71 );
72
73 // Pre-compute grafo:
74 // - `consumers[i]` = índices de edges salientes de `i`.
75 // - `predecessors[j]` = índices de edges entrantes a `j`.
76 let mut consumers: Vec<Vec<usize>> = vec![Vec::new(); n];
77 let mut predecessors: Vec<Vec<usize>> = vec![Vec::new(); n];
78 for (idx, e) in spec.edges.iter().enumerate() {
79 consumers[e.from].push(idx);
80 predecessors[e.to].push(idx);
81 }
82
83 // Por cada edge: par (r_to_consumer, w_from_producer_side).
84 // El consumer recibe r_to_consumer; el producer escribe a w_from_producer_side
85 // (directa o vía splitter).
86 let mut edge_r: Vec<RawFd> = vec![-1; spec.edges.len()];
87 let mut edge_w: Vec<RawFd> = vec![-1; spec.edges.len()];
88 for i in 0..spec.edges.len() {
89 let (r, w) = pipe2(OFlag::O_CLOEXEC).map_err(|e| {
90 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
91 })?;
92 edge_r[i] = r.into_raw_fd();
93 edge_w[i] = w.into_raw_fd();
94 }
95
96 let mut consumer_stdin_fd: Vec<Option<RawFd>> = vec![None; n];
97 let mut producer_stdout_fd: Vec<Option<RawFd>> = vec![None; n];
98 let mut splitter_specs: Vec<SplitterSpec> = Vec::new();
99 let mut merger_specs: Vec<MergerSpec> = Vec::new();
100
101 // Stdout del producer: directo a edge_w[único] si tiene 1 consumer y NO tap;
102 // sino, pipe propio que va al splitter task.
103 for i in 0..n {
104 if consumers[i].is_empty() {
105 continue;
106 }
107 if consumers[i].len() == 1 && !tap {
108 producer_stdout_fd[i] = Some(edge_w[consumers[i][0]]);
109 continue;
110 }
111 // Splitter: pipe propio para el productor → splitter lee y replica a edge_w[*].
112 let (prod_r, prod_w) = pipe2(OFlag::O_CLOEXEC).map_err(|e| {
113 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
114 })?;
115 producer_stdout_fd[i] = Some(prod_w.into_raw_fd());
116 let prod_r_fd = prod_r.into_raw_fd();
117 let mut consumer_writes: Vec<RawFd> = Vec::with_capacity(consumers[i].len());
118 let mut edge_meta: Vec<EdgeMeta> = Vec::with_capacity(consumers[i].len());
119 for edge_idx in &consumers[i] {
120 let edge = &spec.edges[*edge_idx];
121 consumer_writes.push(edge_w[*edge_idx]);
122 edge_meta.push(EdgeMeta {
123 from_label: spec.nodes[edge.from].label.clone(),
124 from_output: edge.from_output.clone(),
125 to_label: spec.nodes[edge.to].label.clone(),
126 to_input: edge.to_input.clone(),
127 });
128 }
129 splitter_specs.push(SplitterSpec {
130 producer_r_fd: prod_r_fd,
131 consumer_w_fds: consumer_writes,
132 edges: edge_meta,
133 tap,
134 sample_bytes: spec.discern.sample_bytes,
135 });
136 }
137
138 // Stdin del consumer: edge_r[único] si tiene 1 predecessor; sino, merger.
139 for j in 0..n {
140 match predecessors[j].len() {
141 0 => {}
142 1 => {
143 consumer_stdin_fd[j] = Some(edge_r[predecessors[j][0]]);
144 }
145 _ => {
146 // Merger: lee de N edge_r y escribe a un nuevo pipe cuyo
147 // read end es el stdin del consumer.
148 let (cons_r, cons_w) = pipe2(OFlag::O_CLOEXEC).map_err(|e| {
149 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
150 })?;
151 consumer_stdin_fd[j] = Some(cons_r.into_raw_fd());
152 let inputs: Vec<RawFd> = predecessors[j]
153 .iter()
154 .map(|eidx| edge_r[*eidx])
155 .collect();
156 merger_specs.push(MergerSpec {
157 producer_r_fds: inputs,
158 consumer_w_fd: cons_w.into_raw_fd(),
159 });
160 }
161 }
162 }
163
164 // Encarnamos cada nodo con su stdin/stdout fd asignado.
165 let mut pids = Vec::with_capacity(n);
166 for (i, node) in spec.nodes.iter().enumerate() {
167 match &node.payload {
168 Payload::Native { .. } | Payload::Legacy { .. } => {}
169 _ => {
170 return Err(CoreError::Incarnate(
171 ente_incarnate::IncarnateError::NonExecutablePayload,
172 ))
173 }
174 }
175 let card = node.to_card(i, workspace_label)?;
176 let stdio = ChildStdio {
177 stdin_fd: consumer_stdin_fd[i],
178 stdout_fd: producer_stdout_fd[i],
179 stderr_fd: None,
180 };
181 let outcome = incarnator
182 .incarnate_with(&card, stdio)
183 .map_err(CoreError::Incarnate)?;
184 let pid = outcome.pid;
185 pids.push((node.label.clone(), pid.as_raw()));
186 debug!(label = %node.label, pid = pid.as_raw(), "node incarnated");
187 }
188
189 let pipeline_id_for_flows = Ulid::new();
190 // Si tap=true, creamos un FlowChannel por edge para el data plane.
191 // Cada splitter pushea al sender del channel correspondiente.
192 let pipeline_id = pipeline_id_for_flows;
193 let mut flow_channels: Vec<crate::flow_channel::FlowChannel> = Vec::new();
194 let mut splitter_channels: Vec<Vec<Option<crate::flow_channel::FlowSender>>> =
195 Vec::with_capacity(splitter_specs.len());
196 let mut edge_socket_for_splitter: Vec<Vec<Option<std::path::PathBuf>>> = Vec::new();
197 for s in &splitter_specs {
198 let mut senders_per_edge = Vec::with_capacity(s.edges.len());
199 let mut paths_per_edge = Vec::with_capacity(s.edges.len());
200 for (i, em) in s.edges.iter().enumerate() {
201 if !s.tap {
202 senders_per_edge.push(None);
203 paths_per_edge.push(None);
204 continue;
205 }
206 let id = format!(
207 "{}-{}-{}-{}",
208 short_ulid(&pipeline_id),
209 em.from_label,
210 em.from_output,
211 i
212 );
213 let socket = crate::flow_channel::default_flow_socket_path(&id);
214 match crate::flow_channel::FlowChannel::with_replay_caps(
215 socket.clone(),
216 crate::flow_channel::ReplayCaps::new(spec.discern.replay_chunks, spec.discern.replay_bytes),
217 ) {
218 Ok(fc) => {
219 senders_per_edge.push(Some(fc.sender_handle()));
220 paths_per_edge.push(Some(socket));
221 flow_channels.push(fc);
222 }
223 Err(e) => {
224 warn!(?e, "flow channel new failed");
225 senders_per_edge.push(None);
226 paths_per_edge.push(None);
227 }
228 }
229 }
230 splitter_channels.push(senders_per_edge);
231 edge_socket_for_splitter.push(paths_per_edge);
232 }
233
234 // Registramos los flow_channels en el manager AHORA, antes de await
235 // las tasks. Esto permite que clientes externos hagan `flow list` y
236 // se suscriban mientras el pipeline aún produce data.
237 if let Some(mgr) = &manager {
238 if !flow_channels.is_empty() {
239 let drained: Vec<crate::flow_channel::FlowChannel> = flow_channels.drain(..).collect();
240 mgr.retain_pipeline_flows(pipeline_id, drained).await;
241 }
242 }
243
244 // Spawn mergers + splitters después del incarnate. Cada task posee
245 // sus fds y los cierra al terminar (via Drop de OwnedFd).
246 let mut merger_handles: Vec<tokio::task::JoinHandle<()>> = Vec::new();
247 for m in merger_specs {
248 merger_handles.push(spawn_merger(m));
249 }
250 let mut tap_handles: Vec<SplitterHandle> = Vec::new();
251 for (s, senders) in splitter_specs.into_iter().zip(splitter_channels.into_iter()) {
252 tap_handles.push(spawn_splitter(s, discerner.clone(), senders));
253 }
254
255 let mut edge_discernments = Vec::new();
256 for (h, paths) in tap_handles.into_iter().zip(edge_socket_for_splitter.into_iter()) {
257 match h.handle.await {
258 Ok(eds) => {
259 for (mut ed, path) in eds.into_iter().zip(paths.into_iter()) {
260 ed.flow_socket = path;
261 edge_discernments.push(ed);
262 }
263 }
264 Err(e) => warn!(?e, "splitter handle joined with error"),
265 }
266 }
267 for h in merger_handles {
268 if let Err(e) = h.await {
269 warn!(?e, "merger handle joined with error");
270 }
271 }
272
273 Ok(PipelineLaunch {
274 pipeline: pipeline_id,
275 command_pids: pids,
276 edge_discernments,
277 })
278}
279
280fn short_ulid(u: &Ulid) -> String {
281 let s = u.to_string();
282 s[s.len() - 6..].to_string()
283}
284
285#[derive(Debug, Clone)]
286struct EdgeMeta {
287 from_label: String,
288 from_output: String,
289 to_label: String,
290 to_input: String,
291}
292
293struct SplitterSpec {
294 producer_r_fd: RawFd,
295 consumer_w_fds: Vec<RawFd>,
296 edges: Vec<EdgeMeta>,
297 tap: bool,
298 sample_bytes: usize,
299}
300
301struct SplitterHandle {
302 handle: tokio::task::JoinHandle<Vec<EdgeDiscernment>>,
303}
304
305struct MergerSpec {
306 producer_r_fds: Vec<RawFd>,
307 consumer_w_fd: RawFd,
308}
309
310fn spawn_merger(spec: MergerSpec) -> tokio::task::JoinHandle<()> {
311 for fd in &spec.producer_r_fds {
312 set_nonblocking(*fd);
313 }
314 set_nonblocking(spec.consumer_w_fd);
315 // Patrón: una task lectora por cada producer reenvía bytes a un mpsc.
316 // El merger principal consume del mpsc y escribe al consumer.
317 // Esto evita el "block en reader idle" del enfoque round-robin sobre
318 // AsyncFd::ready() (los readers idle nunca dejan turno).
319 tokio::spawn(async move {
320 let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<u8>>(32);
321 let nr = spec.producer_r_fds.len();
322 for fd in spec.producer_r_fds {
323 let tx = tx.clone();
324 tokio::spawn(async move {
325 // SAFETY: ownership transferida.
326 let owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(fd) };
327 let r = match AsyncFd::with_interest(owned, Interest::READABLE) {
328 Ok(a) => a,
329 Err(e) => {
330 warn!(?e, "merger reader AsyncFd");
331 return;
332 }
333 };
334 let mut buf = [0u8; 4096];
335 loop {
336 match async_read(&r, &mut buf).await {
337 Ok(0) => break,
338 Ok(n) => {
339 if tx.send(buf[..n].to_vec()).await.is_err() {
340 break;
341 }
342 }
343 Err(_) => break,
344 }
345 }
346 // Drop de tx → cuando todos los readers cerraron, el rx
347 // recibe None y el merger termina.
348 });
349 }
350 drop(tx); // sólo los reader tasks tienen sus clones ahora.
351
352 // SAFETY: ownership transferida al task.
353 let w_owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(spec.consumer_w_fd) };
354 let w = match AsyncFd::with_interest(w_owned, Interest::WRITABLE) {
355 Ok(a) => a,
356 Err(e) => {
357 warn!(?e, "merger AsyncFd w");
358 return;
359 }
360 };
361
362 let mut total: u64 = 0;
363 while let Some(chunk) = rx.recv().await {
364 if async_write_all(&w, &chunk).await.is_err() {
365 return;
366 }
367 total += chunk.len() as u64;
368 }
369 debug!(bytes = total, readers = nr, "merger finished");
370 })
371}
372
373fn spawn_splitter(
374 spec: SplitterSpec,
375 discerner: Arc<DiscernPipeline>,
376 edge_senders: Vec<Option<crate::flow_channel::FlowSender>>,
377) -> SplitterHandle {
378 set_nonblocking(spec.producer_r_fd);
379 for fd in &spec.consumer_w_fds {
380 set_nonblocking(*fd);
381 }
382
383 let handle = tokio::spawn(async move {
384 // SAFETY: ownership transferida al task.
385 let r_owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(spec.producer_r_fd) };
386 let r = match AsyncFd::with_interest(r_owned, Interest::READABLE) {
387 Ok(a) => a,
388 Err(e) => {
389 warn!(?e, "splitter AsyncFd r");
390 return Vec::new();
391 }
392 };
393 let mut writers: Vec<AsyncFd<std::os::fd::OwnedFd>> = Vec::with_capacity(spec.consumer_w_fds.len());
394 for fd in spec.consumer_w_fds {
395 let owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(fd) };
396 match AsyncFd::with_interest(owned, Interest::WRITABLE) {
397 Ok(a) => writers.push(a),
398 Err(e) => warn!(?e, "splitter AsyncFd w"),
399 }
400 }
401
402 let mut sample: Vec<u8> = Vec::with_capacity(spec.sample_bytes);
403 let mut buf = [0u8; 4096];
404 let mut total: u64 = 0;
405 let mut eof = false;
406
407 // Fase 1: sampling (sólo si tap=true) + replicación.
408 while !eof && (spec.tap && sample.len() < spec.sample_bytes) {
409 let n = match async_read(&r, &mut buf).await {
410 Ok(0) => { eof = true; 0 }
411 Ok(n) => n,
412 Err(e) => { warn!(?e, "splitter read"); break; }
413 };
414 if n == 0 { break; }
415 if spec.tap {
416 let take = n.min(spec.sample_bytes - sample.len());
417 sample.extend_from_slice(&buf[..take]);
418 }
419 broadcast_chunk(&writers, &edge_senders, &buf[..n]).await;
420 total += n as u64;
421 }
422
423 let d = if spec.tap {
424 discerner.discern(&sample, &Hint { path: None, size_total: None })
425 } else {
426 None
427 };
428
429 // Fase 2: replicación pura.
430 while !eof {
431 let n = match async_read(&r, &mut buf).await {
432 Ok(0) => { eof = true; 0 }
433 Ok(n) => n,
434 Err(_) => break,
435 };
436 if n == 0 { break; }
437 broadcast_chunk(&writers, &edge_senders, &buf[..n]).await;
438 total += n as u64;
439 }
440 debug!(bytes = total, consumers = writers.len(), "splitter finished");
441
442 // Mismo discernment para todos los edges del splitter (es el mismo
443 // stream replicado). Devolvemos N entries (una por edge) para que
444 // la UI/CLI los liste todos. flow_socket lo rellena el caller.
445 spec.edges
446 .into_iter()
447 .map(|em| EdgeDiscernment {
448 from_label: em.from_label,
449 from_output: em.from_output,
450 to_label: em.to_label,
451 to_input: em.to_input,
452 discernment: d.clone(),
453 flow_socket: None,
454 })
455 .collect()
456 });
457 SplitterHandle { handle }
458}
459
460async fn broadcast_chunk(
461 writers: &[AsyncFd<std::os::fd::OwnedFd>],
462 edge_senders: &[Option<crate::flow_channel::FlowSender>],
463 data: &[u8],
464) {
465 // Internal pipes a los consumers del pipeline.
466 for w in writers {
467 let _ = async_write_all(w, data).await;
468 }
469 // Externos: broadcast a subscribers vía FlowChannel.
470 // Cada edge tiene su propio sender (mismo data — el sample/discernment
471 // viaja por broadcast separados para que un subscriber por edge vea su
472 // stream específico).
473 if edge_senders.iter().any(|s| s.is_some()) {
474 let shared = std::sync::Arc::new(data.to_vec());
475 for s in edge_senders {
476 if let Some(s) = s {
477 let _ = s.send(shared.clone());
478 }
479 }
480 }
481}
482
483async fn async_read(
484 afd: &AsyncFd<std::os::fd::OwnedFd>,
485 buf: &mut [u8],
486) -> std::io::Result<usize> {
487 loop {
488 let mut guard = afd.readable().await?;
489 let fd = afd.as_raw_fd();
490 // SAFETY: lectura sobre fd válido propiedad del AsyncFd.
491 let r = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
492 if r >= 0 {
493 return Ok(r as usize);
494 }
495 let err = std::io::Error::last_os_error();
496 if err.kind() == std::io::ErrorKind::WouldBlock {
497 guard.clear_ready();
498 continue;
499 }
500 return Err(err);
501 }
502}
503
504async fn async_write_all(
505 afd: &AsyncFd<std::os::fd::OwnedFd>,
506 mut buf: &[u8],
507) -> std::io::Result<()> {
508 while !buf.is_empty() {
509 let mut guard = afd.writable().await?;
510 let fd = afd.as_raw_fd();
511 // SAFETY: escritura sobre fd válido propiedad del AsyncFd.
512 let r = unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) };
513 if r > 0 {
514 buf = &buf[r as usize..];
515 continue;
516 }
517 if r == 0 {
518 return Err(std::io::Error::new(
519 std::io::ErrorKind::WriteZero,
520 "write 0",
521 ));
522 }
523 let err = std::io::Error::last_os_error();
524 if err.kind() == std::io::ErrorKind::WouldBlock {
525 guard.clear_ready();
526 continue;
527 }
528 return Err(err);
529 }
530 Ok(())
531}
532
533fn set_nonblocking(fd: RawFd) {
534 // SAFETY: fcntl con F_SETFL es seguro para fds válidos.
535 unsafe {
536 let flags = libc::fcntl(fd, libc::F_GETFL, 0);
537 if flags >= 0 {
538 libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
539 }
540 }
541}
542
543// Extension trait para abstraer la API de OwnedFd entre versiones (compat).
544trait OwnedFdFromRawCompat: Sized {
545 unsafe fn from_raw_fd_compat(fd: RawFd) -> Self;
546}
547
548impl OwnedFdFromRawCompat for std::os::fd::OwnedFd {
549 unsafe fn from_raw_fd_compat(fd: RawFd) -> Self {
550 use std::os::fd::FromRawFd;
551 // SAFETY: el caller transfiere ownership de `fd` a la `OwnedFd`.
552 unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }
553 }
554}
555
556// Re-export para que el unused warning del AsRawFd se calle si no se usa.
557#[allow(dead_code)]
558fn _keep_raw(_: &dyn AsRawFd) {}
559
560#[cfg(test)]
561mod tests {
562 use super::*;
563 use brahman_card::Payload;
564 use ente_incarnate::IncarnatorConfig;
565 use shipote_card::{CommandRef, DiscernPolicy, FlowEdge, PipelineSpec, WorkspaceId};
566
567 fn cmd(label: &str, exec: &str, argv: &[&str]) -> CommandRef {
568 CommandRef {
569 label: label.into(),
570 payload: Payload::Native {
571 exec: exec.into(),
572 argv: argv.iter().map(|s| s.to_string()).collect(),
573 envp: vec![],
574 },
575 soma: Default::default(),
576 flows: Default::default(),
577 supervision: brahman_card::Supervision::OneShot,
578 }
579 }
580
581 #[tokio::test]
582 async fn pipeline_isolated_echo_to_cat_runs() {
583 let spec = PipelineSpec {
584 label: "echo-cat".into(),
585 workspace: WorkspaceId::new(),
586 nodes: vec![
587 cmd("p1", "/bin/echo", &["hola pipeline aislado"]),
588 cmd("p2", "/bin/cat", &[]),
589 ],
590 edges: vec![FlowEdge {
591 from: 0,
592 from_output: "stdout".into(),
593 to: 1,
594 to_input: "stdin".into(),
595 }],
596 discern: DiscernPolicy::default(),
597 };
598 let disc = Arc::new(DiscernPipeline::default_pipeline());
599 let inc = Arc::new(Incarnator::new(IncarnatorConfig::default()));
600 let launch = run_pipeline(&spec, "ws", false, disc, inc, None).await.unwrap();
601 assert_eq!(launch.command_pids.len(), 2);
602 // Cosecha.
603 for (_, pid) in &launch.command_pids {
604 let _ = nix::sys::wait::waitpid(nix::unistd::Pid::from_raw(*pid), None);
605 }
606 }
607
608 #[tokio::test]
609 async fn pipeline_fanin_two_to_one() {
610 // 2 productores → 1 consumer (cat). El merger multiplexa.
611 let spec = PipelineSpec {
612 label: "fanin".into(),
613 workspace: WorkspaceId::new(),
614 nodes: vec![
615 cmd("p1", "/bin/echo", &["from-p1"]),
616 cmd("p2", "/bin/echo", &["from-p2"]),
617 cmd("c", "/bin/cat", &[]),
618 ],
619 edges: vec![
620 FlowEdge {
621 from: 0,
622 from_output: "stdout".into(),
623 to: 2,
624 to_input: "stdin".into(),
625 },
626 FlowEdge {
627 from: 1,
628 from_output: "stdout".into(),
629 to: 2,
630 to_input: "stdin".into(),
631 },
632 ],
633 discern: DiscernPolicy::default(),
634 };
635 let disc = Arc::new(DiscernPipeline::default_pipeline());
636 let inc = Arc::new(Incarnator::new(IncarnatorConfig::default()));
637 let launch = run_pipeline(&spec, "ws", false, disc, inc, None).await.unwrap();
638 assert_eq!(launch.command_pids.len(), 3);
639 for (_, pid) in &launch.command_pids {
640 let _ = nix::sys::wait::waitpid(nix::unistd::Pid::from_raw(*pid), None);
641 }
642 }
643
644 #[tokio::test]
645 async fn pipeline_fanout_one_to_two() {
646 // 1 productor (echo) → 2 consumers (wc -c). Splitter replica.
647 let spec = PipelineSpec {
648 label: "fanout".into(),
649 workspace: WorkspaceId::new(),
650 nodes: vec![
651 cmd("p", "/bin/echo", &["fanout-test"]),
652 cmd("c1", "/bin/cat", &[]),
653 cmd("c2", "/bin/cat", &[]),
654 ],
655 edges: vec![
656 FlowEdge {
657 from: 0,
658 from_output: "stdout".into(),
659 to: 1,
660 to_input: "stdin".into(),
661 },
662 FlowEdge {
663 from: 0,
664 from_output: "stdout".into(),
665 to: 2,
666 to_input: "stdin".into(),
667 },
668 ],
669 discern: DiscernPolicy::default(),
670 };
671 let disc = Arc::new(DiscernPipeline::default_pipeline());
672 let inc = Arc::new(Incarnator::new(IncarnatorConfig::default()));
673 let launch = run_pipeline(&spec, "ws", false, disc, inc, None).await.unwrap();
674 assert_eq!(launch.command_pids.len(), 3);
675 for (_, pid) in &launch.command_pids {
676 let _ = nix::sys::wait::waitpid(nix::unistd::Pid::from_raw(*pid), None);
677 }
678 }
679
680 #[tokio::test]
681 async fn pipeline_isolated_with_tap_captures_discernment() {
682 let spec = PipelineSpec {
683 label: "json-cat".into(),
684 workspace: WorkspaceId::new(),
685 nodes: vec![
686 cmd("p1", "/bin/echo", &["{\"hello\": 1}"]),
687 cmd("p2", "/bin/cat", &[]),
688 ],
689 edges: vec![FlowEdge {
690 from: 0,
691 from_output: "stdout".into(),
692 to: 1,
693 to_input: "stdin".into(),
694 }],
695 discern: DiscernPolicy {
696 sample_bytes: 4096,
697 enrich_producer: true,
698 replay_chunks: 32,
699 replay_bytes: 0,
700 },
701 };
702 let disc = Arc::new(DiscernPipeline::default_pipeline());
703 let inc = Arc::new(Incarnator::new(IncarnatorConfig::default()));
704 let launch = run_pipeline(&spec, "ws", true, disc, inc, None).await.unwrap();
705 assert_eq!(launch.edge_discernments.len(), 1);
706 let d = &launch.edge_discernments[0];
707 let dis = d.discernment.as_ref().expect("discernment present");
708 assert_eq!(dis.mime.as_deref(), Some("application/json"));
709 // Cosecha.
710 for (_, pid) in &launch.command_pids {
711 let _ = nix::sys::wait::waitpid(nix::unistd::Pid::from_raw(*pid), None);
712 }
713 }
714}