⎇ 4c9d1b4c1dbc62cf7d9cfde211524eb06ba21144
crates/modules/shipote/shipote-core/src/lib.rs
History
Who touched what
raw
feat(shipote): quota enforce + cgroup memory.max + pipeline restart (fase L)
1//! `shipote-core` — runtime in-memory de Workspaces y comandos.
2//!
3//! Mantiene un estado tokio-friendly (Mutex sobre HashMap) con:
4//! - Workspaces vivos (id → state).
5//! - PIDs de comandos lanzados, indexados por workspace.
6//! - Reaping cooperativo: `reap_dead()` cosecha hijos terminados.
7
8// `pipeline` necesita `unsafe` puntual para `libc::close` y construir
9// `OwnedFd` desde fds que armamos con `pipe2(2)`. El resto del crate
10// permanece safe — el cargo lint `unsafe_code` queda permitido sólo en
11// el módulo concreto.
12#![deny(unsafe_op_in_unsafe_fn)]
13
14pub mod flow_channel;
15pub mod logbuf;
16pub mod persist;
17pub mod pipeline;
18pub mod stats;
19
20use brahman_card::{Card, Payload, Supervision};
21use ente_incarnate::{Incarnator, IncarnatorConfig};
22use nix::sys::signal::{kill, Signal};
23use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
24use nix::unistd::Pid;
25use shipote_card::{CommandRef, PipelineSpec, WorkspaceId, WorkspaceSpec};
26use std::collections::HashMap;
27use std::sync::Arc;
28use std::time::Instant;
29use thiserror::Error;
30use tokio::sync::Mutex;
31use tracing::{info, warn};
32use ulid::Ulid;
33
34#[derive(Debug, Error)]
35pub enum CoreError {
36 #[error("workspace {0} not found")]
37 WorkspaceNotFound(WorkspaceId),
38 #[error("compile: {0}")]
39 Compile(#[from] shipote_card::CompileError),
40 #[error("incarnate: {0}")]
41 Incarnate(#[from] ente_incarnate::IncarnateError),
42}
43
44#[derive(Debug)]
45pub struct WorkspaceState {
46 pub id: WorkspaceId,
47 pub spec: WorkspaceSpec,
48 pub root_card: Card,
49 pub commands: HashMap<Ulid, CommandState>,
50 pub started: Instant,
51 /// Última muestra de `(wall_instant, cpu_usec)` usada para calcular
52 /// `cpu_percent` en la próxima medición. None hasta el primer measure.
53 pub last_cpu_sample: Option<(Instant, u64)>,
54}
55
56#[derive(Debug, Clone)]
57pub struct CommandState {
58 pub id: Ulid,
59 pub label: String,
60 pub pid: Pid,
61 pub alive: bool,
62 pub exit_status: Option<i32>,
63 /// Ring buffer del stdout. `None` para comandos sin captura.
64 pub stdout: Option<logbuf::LogBuf>,
65 /// Ring buffer del stderr. Separado de `stdout` para que el CLI
66 /// pueda filtrarlos. `None` para comandos sin captura.
67 pub stderr: Option<logbuf::LogBuf>,
68 /// Si el comando fue lanzado como parte de un Pipeline, su ULID.
69 pub pipeline_id: Option<Ulid>,
70}
71
72/// Stream a leer en `get_command_logs`. `Both` concatena stderr-después-stdout
73/// para una vista combinada (orden temporal aproximado).
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub enum LogStream {
76 Stdout,
77 Stderr,
78 Both,
79}
80
81pub struct WorkspaceManager {
82 inner: Arc<Mutex<Inner>>,
83 incarnator: Arc<Incarnator>,
84}
85
86struct Inner {
87 workspaces: HashMap<WorkspaceId, WorkspaceState>,
88 /// Definiciones nombradas de pipelines persistidas. NO es lo mismo
89 /// que "pipelines vivos" — son specs guardados para reusar con
90 /// `run-saved`. Sobreviven restart vía snapshot.
91 saved_pipelines: HashMap<String, PipelineSpec>,
92 /// Flow channels vivos por pipeline. Se retienen hasta que el
93 /// pipeline termine — cuando todos los hijos del pipeline murieron,
94 /// el reaper los borra (futuro). v1: viven hasta `stop_pipeline_flows`
95 /// explícito o hasta shutdown.
96 pipeline_flows: HashMap<Ulid, Vec<crate::flow_channel::FlowChannel>>,
97 /// Specs de comandos `run()` con `restart_on_failure=true`. Indexed
98 /// por command_id. Cuando `reap_dead` detecta exit!=0, se relauncha
99 /// con la misma spec (nuevo pid y nuevo command_id se asigna por
100 /// el nuevo state pero el restart_spec sigue ligado al original).
101 restart_specs: HashMap<Ulid, RestartSpec>,
102 /// Supervisores de pipelines con `restart_on_failure`. Indexed por
103 /// pipeline_id. Cuando `reap_dead` detecta que el pipeline tuvo
104 /// algún command failure, agrega un entry a `pending_pipeline_restarts`.
105 pipeline_supervisors: HashMap<Ulid, PipelineSupervisor>,
106 /// Cola de pipelines pendientes de restart. El daemon la drena en
107 /// cada loop del reaper, hace stop + run_pipeline.
108 pending_pipeline_restarts: Vec<Ulid>,
109}
110
111#[derive(Debug, Clone)]
112pub struct PipelineSupervisor {
113 pub workspace: WorkspaceId,
114 pub spec: PipelineSpec,
115 pub tap: bool,
116 pub restart_count: u32,
117}
118
119#[derive(Debug, Clone)]
120struct RestartSpec {
121 workspace: WorkspaceId,
122 exec: String,
123 argv: Vec<String>,
124 envp: Vec<(String, String)>,
125 /// Backoff inicial (ms). Crece exponencialmente hasta max_backoff_ms.
126 backoff_ms: u64,
127 max_backoff_ms: u64,
128 /// Cantidad de restarts ya ejecutados (para tracking).
129 restart_count: u32,
130}
131
132#[derive(Debug, Clone)]
133pub struct CommandSummary {
134 pub id: Ulid,
135 pub label: String,
136 pub pid: i32,
137}
138
139#[derive(Debug, Clone)]
140pub struct CommandInfo {
141 pub id: Ulid,
142 pub label: String,
143 pub pid: i32,
144 pub alive: bool,
145 pub exit_status: Option<i32>,
146 pub log_bytes: u64,
147}
148
149/// Lee VmRSS (bytes) de `/proc/<pid>/status`. Helper local para
150/// reap_dead que no necesita el full stats. Devuelve 0 si el proc no
151/// existe o el campo no aparece.
152fn read_proc_rss(pid: i32) -> Option<u64> {
153 let status = std::fs::read_to_string(format!("/proc/{pid}/status")).ok()?;
154 status
155 .lines()
156 .find_map(|l| l.strip_prefix("VmRSS:").map(str::trim))
157 .and_then(|s| s.split_whitespace().next())
158 .and_then(|s| s.parse::<u64>().ok())
159 .map(|kb| kb * 1024)
160}
161
162fn spawn_log_drainer(read_fd: std::os::fd::RawFd, logs: logbuf::LogBuf) {
163 // Marcar non-blocking + envolver en AsyncFd; igual patrón que el tap.
164 // SAFETY: F_SETFL sobre fd válido.
165 unsafe {
166 let flags = libc::fcntl(read_fd, libc::F_GETFL, 0);
167 if flags >= 0 {
168 libc::fcntl(read_fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
169 }
170 }
171 tokio::spawn(async move {
172 // SAFETY: ownership del fd transferido al drainer task.
173 let owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(read_fd) };
174 let afd = match tokio::io::unix::AsyncFd::with_interest(owned, tokio::io::Interest::READABLE) {
175 Ok(a) => a,
176 Err(e) => {
177 tracing::warn!(?e, "log drainer AsyncFd failed");
178 return;
179 }
180 };
181 let mut buf = [0u8; 4096];
182 loop {
183 let mut guard = match afd.readable().await {
184 Ok(g) => g,
185 Err(_) => break,
186 };
187 use std::os::fd::AsRawFd;
188 let fd = afd.as_raw_fd();
189 // SAFETY: read sobre fd válido.
190 let r = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
191 if r > 0 {
192 logs.append(&buf[..r as usize]);
193 continue;
194 }
195 if r == 0 {
196 break; // EOF
197 }
198 let err = std::io::Error::last_os_error();
199 if err.kind() == std::io::ErrorKind::WouldBlock {
200 guard.clear_ready();
201 continue;
202 }
203 tracing::warn!(?err, "log drainer read err");
204 break;
205 }
206 });
207}
208
209trait OwnedFdFromRawCompat: Sized {
210 unsafe fn from_raw_fd_compat(fd: std::os::fd::RawFd) -> Self;
211}
212
213impl OwnedFdFromRawCompat for std::os::fd::OwnedFd {
214 unsafe fn from_raw_fd_compat(fd: std::os::fd::RawFd) -> Self {
215 use std::os::fd::FromRawFd;
216 // SAFETY: el caller transfiere ownership de fd a OwnedFd.
217 unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }
218 }
219}
220
221impl WorkspaceManager {
222 pub fn new(cfg: IncarnatorConfig) -> Self {
223 Self {
224 inner: Arc::new(Mutex::new(Inner {
225 workspaces: HashMap::new(),
226 saved_pipelines: HashMap::new(),
227 pipeline_flows: HashMap::new(),
228 restart_specs: HashMap::new(),
229 pipeline_supervisors: HashMap::new(),
230 pending_pipeline_restarts: Vec::new(),
231 })),
232 incarnator: Arc::new(Incarnator::new(cfg)),
233 }
234 }
235
236 /// Registra un supervisor para un pipeline con `restart_on_failure=true`.
237 /// El daemon llama esto tras `run_pipeline` para que `reap_dead` agregue
238 /// el pipeline a la cola de restart cuando algún command falle.
239 pub async fn register_pipeline_supervisor(
240 &self,
241 pipeline_id: Ulid,
242 workspace: WorkspaceId,
243 spec: PipelineSpec,
244 tap: bool,
245 ) {
246 if !spec.restart_on_failure {
247 return;
248 }
249 tracing::debug!(%pipeline_id, label = %spec.label, "pipeline supervisor registered");
250 let mut g = self.inner.lock().await;
251 g.pipeline_supervisors.insert(
252 pipeline_id,
253 PipelineSupervisor {
254 workspace,
255 spec,
256 tap,
257 restart_count: 0,
258 },
259 );
260 }
261
262 /// Drena la cola de pipelines pendientes de restart y retorna las
263 /// specs a relaunch. El daemon lo llama tras cada `reap_dead`.
264 pub async fn take_pending_restarts(&self) -> Vec<PipelineSupervisor> {
265 let mut g = self.inner.lock().await;
266 let pending = std::mem::take(&mut g.pending_pipeline_restarts);
267 let mut out = Vec::with_capacity(pending.len());
268 for old_id in pending {
269 if let Some(mut sup) = g.pipeline_supervisors.remove(&old_id) {
270 sup.restart_count += 1;
271 out.push(sup);
272 }
273 }
274 out
275 }
276
277 /// Registra los comandos lanzados por un pipeline en el workspace.
278 /// Esto permite `pipeline_stop` (matar selectivamente sólo los pids
279 /// de un pipeline). `pipeline_id` se setea en cada CommandState.
280 pub async fn register_pipeline_commands(
281 &self,
282 workspace: WorkspaceId,
283 pipeline_id: Ulid,
284 commands: Vec<(String, i32)>,
285 ) {
286 let mut g = self.inner.lock().await;
287 let Some(ws) = g.workspaces.get_mut(&workspace) else { return };
288 for (label, pid) in commands {
289 let cmd_id = Ulid::new();
290 ws.commands.insert(
291 cmd_id,
292 CommandState {
293 id: cmd_id,
294 label,
295 pid: Pid::from_raw(pid),
296 alive: true,
297 exit_status: None,
298 stdout: None,
299 stderr: None,
300 pipeline_id: Some(pipeline_id),
301 },
302 );
303 }
304 }
305
306 /// Detiene selectivamente los comandos de un pipeline. SIGTERM →
307 /// `grace` → SIGKILL. Devuelve cantidad reapeada. Si no hay comandos
308 /// del pipeline en ningún workspace, retorna 0.
309 pub async fn stop_pipeline(
310 &self,
311 pipeline_id: Ulid,
312 grace: std::time::Duration,
313 ) -> u32 {
314 // 1) Recolectamos pids de ese pipeline en todos los workspaces.
315 let mut targets: Vec<Pid> = Vec::new();
316 {
317 let g = self.inner.lock().await;
318 for ws in g.workspaces.values() {
319 for cmd in ws.commands.values() {
320 if cmd.alive && cmd.pipeline_id == Some(pipeline_id) {
321 targets.push(cmd.pid);
322 }
323 }
324 }
325 }
326 if targets.is_empty() {
327 return 0;
328 }
329 let initial = if grace.is_zero() { Signal::SIGKILL } else { Signal::SIGTERM };
330 for pid in &targets {
331 let _ = kill(*pid, initial);
332 }
333 let mut reaped = 0u32;
334 let mut still = targets.clone();
335 let deadline = std::time::Instant::now() + grace;
336 let poll = std::time::Duration::from_millis(20);
337 while !still.is_empty() && std::time::Instant::now() < deadline {
338 still.retain(|pid| match waitpid(*pid, Some(WaitPidFlag::WNOHANG)) {
339 Ok(WaitStatus::StillAlive) => true,
340 Ok(_) => {
341 reaped += 1;
342 false
343 }
344 Err(_) => false,
345 });
346 if !still.is_empty() {
347 tokio::time::sleep(poll).await;
348 }
349 }
350 for pid in &still {
351 let _ = kill(*pid, Signal::SIGKILL);
352 let _ = waitpid(*pid, None);
353 reaped += 1;
354 }
355 // Marcar como dead en estado in-memory.
356 let mut g = self.inner.lock().await;
357 for ws in g.workspaces.values_mut() {
358 for cmd in ws.commands.values_mut() {
359 if cmd.pipeline_id == Some(pipeline_id) && cmd.alive {
360 cmd.alive = false;
361 }
362 }
363 }
364 // Drop flows del pipeline.
365 g.pipeline_flows.remove(&pipeline_id);
366 info!(%pipeline_id, reaped, "pipeline stopped");
367 reaped
368 }
369
370 /// Retiene los FlowChannels de un pipeline para que sobrevivan al
371 /// fin del request. Drop = cierre del data plane.
372 pub async fn retain_pipeline_flows(
373 &self,
374 pipeline: Ulid,
375 flows: Vec<crate::flow_channel::FlowChannel>,
376 ) {
377 self.inner.lock().await.pipeline_flows.insert(pipeline, flows);
378 }
379
380 /// Lista pipelines vivos con sus sockets activos.
381 pub async fn list_flow_pipelines(&self) -> Vec<(Ulid, Vec<std::path::PathBuf>)> {
382 let g = self.inner.lock().await;
383 g.pipeline_flows
384 .iter()
385 .map(|(id, flows)| {
386 (
387 *id,
388 flows.iter().map(|f| f.socket_path().to_path_buf()).collect(),
389 )
390 })
391 .collect()
392 }
393
394 /// Cierra el data plane de un pipeline (drop = remove_file de sockets).
395 pub async fn drop_pipeline_flows(&self, pipeline: Ulid) -> bool {
396 self.inner.lock().await.pipeline_flows.remove(&pipeline).is_some()
397 }
398
399 pub fn incarnator(&self) -> &Incarnator {
400 &self.incarnator
401 }
402
403 /// Handle Arc-clonable del Incarnator, para que el pipeline lo pueda
404 /// usar fuera del manager.
405 pub fn incarnator_handle(&self) -> Arc<Incarnator> {
406 self.incarnator.clone()
407 }
408
409 // -----------------------------------------------------------------
410 // Saved pipelines (definiciones nombradas, no runs)
411 // -----------------------------------------------------------------
412
413 /// Guarda (o reemplaza) un PipelineSpec bajo `name`.
414 pub async fn save_pipeline(&self, name: String, spec: PipelineSpec) {
415 self.inner.lock().await.saved_pipelines.insert(name, spec);
416 }
417
418 /// Devuelve los nombres de los pipelines guardados.
419 pub async fn list_saved_pipelines(&self) -> Vec<String> {
420 let g = self.inner.lock().await;
421 let mut v: Vec<String> = g.saved_pipelines.keys().cloned().collect();
422 v.sort();
423 v
424 }
425
426 /// Recupera el PipelineSpec guardado bajo `name`.
427 pub async fn get_saved_pipeline(&self, name: &str) -> Option<PipelineSpec> {
428 self.inner.lock().await.saved_pipelines.get(name).cloned()
429 }
430
431 /// Elimina un saved pipeline.
432 pub async fn drop_saved_pipeline(&self, name: &str) -> bool {
433 self.inner.lock().await.saved_pipelines.remove(name).is_some()
434 }
435
436 /// Label del workspace, si existe.
437 pub async fn workspace_label(&self, id: WorkspaceId) -> Option<String> {
438 self.inner
439 .lock()
440 .await
441 .workspaces
442 .get(&id)
443 .map(|w| w.spec.label.clone())
444 }
445
446 /// Compara accounting real (RSS, commands_alive) contra los rlimits
447 /// declarados en `SomaSpec`. Devuelve violaciones humanizadas. NO
448 /// hace enforcement automático.
449 pub async fn workspace_quota(&self, id: WorkspaceId) -> Option<stats::QuotaReport> {
450 let stats_now = self.workspace_stats(id).await?;
451 let g = self.inner.lock().await;
452 let ws = g.workspaces.get(&id)?;
453 let rl = &ws.spec.soma.rlimits;
454 let mut report = stats::QuotaReport {
455 mem_limit: rl.mem_bytes,
456 nproc_limit: rl.nproc,
457 breaches: Vec::new(),
458 };
459 if let (Some(limit), Some(used)) = (rl.mem_bytes, stats_now.rss_bytes) {
460 if used > limit {
461 report.breaches.push(format!(
462 "memory: {:.2} MiB > {:.2} MiB limit",
463 used as f64 / 1024.0 / 1024.0,
464 limit as f64 / 1024.0 / 1024.0,
465 ));
466 }
467 }
468 if let Some(limit) = rl.nproc {
469 if stats_now.commands_alive > limit {
470 report.breaches.push(format!(
471 "nproc: {} alive > {} limit",
472 stats_now.commands_alive, limit
473 ));
474 }
475 }
476 Some(report)
477 }
478
479 /// Estadísticas de recursos del workspace: RSS + CPU agregado de sus
480 /// comandos vivos. Lee `/proc/<pid>/` directamente; si el spec declara
481 /// `soma.cgroup.path`, también intenta el cgroup (más preciso, incluye
482 /// descendants).
483 ///
484 /// `cpu_percent` se calcula entre samples consecutivos. Necesita ≥2
485 /// llamadas para tener un valor (la primera siempre retorna `None`).
486 pub async fn workspace_stats(&self, id: WorkspaceId) -> Option<stats::WorkspaceStats> {
487 let mut g = self.inner.lock().await;
488 let ws = g.workspaces.get_mut(&id)?;
489 let alive: Vec<i32> = ws
490 .commands
491 .values()
492 .filter(|c| c.alive)
493 .map(|c| c.pid.as_raw())
494 .collect();
495 let total = ws.commands.len() as u32;
496 let cgroup_path = if ws.spec.soma.cgroup.path.is_empty() {
497 None
498 } else {
499 Some(std::path::PathBuf::from(format!(
500 "/sys/fs/cgroup{}",
501 ws.spec.soma.cgroup.path
502 )))
503 };
504 let mut s = stats::measure(&alive, cgroup_path.as_deref(), ws.started);
505 s.commands_total = total;
506
507 // CPU%: diff entre el sample actual y el previo, dividido por
508 // wall time. 100% = 1 core saturado. >100% = varios cores.
509 let now = Instant::now();
510 if let Some(cpu_now) = s.cpu_usec {
511 if let Some((prev_t, prev_cpu)) = ws.last_cpu_sample {
512 let dt_us = now.duration_since(prev_t).as_micros() as u64;
513 let d_cpu = cpu_now.saturating_sub(prev_cpu);
514 if dt_us > 0 {
515 s.cpu_percent = Some(100.0 * d_cpu as f32 / dt_us as f32);
516 }
517 }
518 ws.last_cpu_sample = Some((now, cpu_now));
519 }
520 Some(s)
521 }
522
523 pub async fn create(
524 self: &Arc<Self>,
525 spec: WorkspaceSpec,
526 ) -> Result<(WorkspaceId, Vec<String>), CoreError> {
527 self.create_with_id(WorkspaceId::new(), spec).await
528 }
529
530 /// Variante que acepta el ID. Útil para restore_snapshot: preserva
531 /// ULIDs entre restarts, así clients que tracking workspace_id no se
532 /// rompen.
533 pub async fn create_with_id(
534 self: &Arc<Self>,
535 id: WorkspaceId,
536 spec: WorkspaceSpec,
537 ) -> Result<(WorkspaceId, Vec<String>), CoreError> {
538 let card = spec.to_card(id)?;
539 let mut warnings = self.incarnator.dry_run(&card).warnings;
540 let ttl = spec.ttl;
541
542 // Si el workspace declara cgroup path Y rlimits, intentamos
543 // crear el cgroup y escribir memory.max/pids.max. El kernel
544 // hace OOM kill al exceder memory.max — enforcement automático
545 // sin policy adicional. Falla silenciosa si no hay delegation.
546 if !spec.soma.cgroup.path.is_empty() {
547 if let Ok(abs) = ente_incarnate::cgroup::ensure_cgroup(&spec.soma.cgroup) {
548 let applied =
549 ente_incarnate::cgroup::apply_rlimits_to_cgroup(&abs, &spec.soma.rlimits);
550 if !applied.is_empty() {
551 warnings.push(format!("cgroup limits applied: {}", applied.join(", ")));
552 }
553 }
554 }
555 let state = WorkspaceState {
556 id,
557 spec,
558 root_card: card,
559 commands: HashMap::new(),
560 started: Instant::now(),
561 last_cpu_sample: None,
562 };
563 self.inner.lock().await.workspaces.insert(id, state);
564 info!(%id, ?ttl, "workspace created");
565
566 // Si tiene TTL, programar auto-stop. El task captura un weak ref
567 // al manager para no impedir que se dropée si el daemon termina.
568 if let Some(duration) = ttl {
569 let mgr_weak = Arc::downgrade(self);
570 tokio::spawn(async move {
571 tokio::time::sleep(duration).await;
572 if let Some(mgr) = mgr_weak.upgrade() {
573 let exists = mgr.inner.lock().await.workspaces.contains_key(&id);
574 if exists {
575 info!(%id, "workspace TTL expired — auto-stop");
576 let _ = mgr.stop(id).await;
577 }
578 }
579 });
580 }
581
582 Ok((id, warnings))
583 }
584
585 pub async fn list(&self) -> Vec<WorkspaceSnapshot> {
586 let g = self.inner.lock().await;
587 g.workspaces
588 .values()
589 .map(|w| WorkspaceSnapshot {
590 id: w.id,
591 label: w.spec.label.clone(),
592 commands: w.commands.len() as u32,
593 uptime_ms: w.started.elapsed().as_millis() as u64,
594 })
595 .collect()
596 }
597
598 pub async fn stop(&self, id: WorkspaceId) -> Result<u32, CoreError> {
599 self.stop_with_grace(id, std::time::Duration::from_millis(1000)).await
600 }
601
602 /// Variante con tiempo de gracia configurable. SIGTERM → espera `grace`
603 /// → SIGKILL si quedan vivos. `grace=0` = SIGKILL inmediato.
604 pub async fn stop_with_grace(
605 &self,
606 id: WorkspaceId,
607 grace: std::time::Duration,
608 ) -> Result<u32, CoreError> {
609 let mut g = self.inner.lock().await;
610 let ws = g.workspaces.remove(&id).ok_or(CoreError::WorkspaceNotFound(id))?;
611 // También limpiamos flow_channels del workspace si los hubiera —
612 // por workspace lo retenemos por pipeline, no por workspace.
613 drop(g);
614
615 // 1) SIGTERM (o SIGKILL si grace=0) a todos vivos.
616 let initial_signal = if grace.is_zero() { Signal::SIGKILL } else { Signal::SIGTERM };
617 let alive_pids: Vec<Pid> = ws
618 .commands
619 .values()
620 .filter(|c| c.alive)
621 .map(|c| c.pid)
622 .collect();
623 for pid in &alive_pids {
624 let _ = kill(*pid, initial_signal);
625 }
626
627 // 2) Esperar hasta `grace` haciendo polling WNOHANG.
628 let mut reaped = 0u32;
629 let mut still_alive: Vec<Pid> = alive_pids.clone();
630 let deadline = std::time::Instant::now() + grace;
631 let poll_interval = std::time::Duration::from_millis(20);
632 while !still_alive.is_empty() && std::time::Instant::now() < deadline {
633 still_alive.retain(|pid| match waitpid(*pid, Some(WaitPidFlag::WNOHANG)) {
634 Ok(WaitStatus::StillAlive) => true,
635 Ok(_) => {
636 reaped += 1;
637 false
638 }
639 Err(_) => false,
640 });
641 if !still_alive.is_empty() {
642 tokio::time::sleep(poll_interval).await;
643 }
644 }
645
646 // 3) SIGKILL forzoso a los que quedan, y wait blocking.
647 for pid in &still_alive {
648 let _ = kill(*pid, Signal::SIGKILL);
649 let _ = waitpid(*pid, None);
650 reaped += 1;
651 }
652 info!(
653 %id,
654 reaped,
655 grace_ms = grace.as_millis() as u64,
656 sigkilled = still_alive.len(),
657 "workspace stopped"
658 );
659 Ok(reaped)
660 }
661
662 /// Ejecuta un comando one-shot dentro de un workspace existente.
663 /// Captura stdout+stderr en un ring buffer accesible vía
664 /// [`get_command_logs`](Self::get_command_logs).
665 pub async fn run(
666 &self,
667 id: WorkspaceId,
668 exec: String,
669 argv: Vec<String>,
670 envp: Vec<(String, String)>,
671 ) -> Result<CommandSummary, CoreError> {
672 self.run_with_options(id, exec, argv, envp, false).await
673 }
674
675 /// Variante con `restart_on_failure`: si el comando muere con
676 /// exit_status != 0, el reaper lo relauncha con backoff exponencial
677 /// (200ms → 400 → 800 → … cap 30s).
678 pub async fn run_with_options(
679 &self,
680 id: WorkspaceId,
681 exec: String,
682 argv: Vec<String>,
683 envp: Vec<(String, String)>,
684 restart_on_failure: bool,
685 ) -> Result<CommandSummary, CoreError> {
686 let workspace_label = {
687 let g = self.inner.lock().await;
688 let ws = g.workspaces.get(&id).ok_or(CoreError::WorkspaceNotFound(id))?;
689 ws.spec.label.clone()
690 };
691 let cmd_ref = CommandRef {
692 label: format!("run-{}", short_ulid(&Ulid::new())),
693 payload: Payload::Native { exec, argv, envp },
694 soma: Default::default(),
695 flows: Default::default(),
696 supervision: Supervision::OneShot,
697 };
698 let card = cmd_ref.to_card(0, &workspace_label)?;
699
700 // Dos pipes O_CLOEXEC: uno para stdout, otro para stderr.
701 use std::os::fd::IntoRawFd;
702 let (sout_r, sout_w) =
703 nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC).map_err(|e| {
704 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
705 })?;
706 let (serr_r, serr_w) =
707 nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC).map_err(|e| {
708 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
709 })?;
710 let sout_r_fd = sout_r.into_raw_fd();
711 let sout_w_fd = sout_w.into_raw_fd();
712 let serr_r_fd = serr_r.into_raw_fd();
713 let serr_w_fd = serr_w.into_raw_fd();
714
715 let stdout_buf = logbuf::LogBuf::new();
716 let stderr_buf = logbuf::LogBuf::new();
717
718 let stdio = ente_incarnate::ChildStdio {
719 stdin_fd: None,
720 stdout_fd: Some(sout_w_fd),
721 stderr_fd: Some(serr_w_fd),
722 };
723 let out = self.incarnator.incarnate_with(&card, stdio)?;
724 let cmd_id = card.id;
725 let cmd_label = cmd_ref.label.clone();
726 let pid = out.pid;
727
728 spawn_log_drainer(sout_r_fd, stdout_buf.clone());
729 spawn_log_drainer(serr_r_fd, stderr_buf.clone());
730
731 let mut g = self.inner.lock().await;
732 if let Some(ws) = g.workspaces.get_mut(&id) {
733 ws.commands.insert(
734 cmd_id,
735 CommandState {
736 id: cmd_id,
737 label: cmd_label.clone(),
738 pid,
739 alive: true,
740 exit_status: None,
741 stdout: Some(stdout_buf),
742 stderr: Some(stderr_buf),
743 pipeline_id: None,
744 },
745 );
746 }
747 if restart_on_failure {
748 // Reextract exec/argv/envp del payload del CommandRef.
749 if let Payload::Native { exec, argv, envp } = &cmd_ref.payload {
750 g.restart_specs.insert(
751 cmd_id,
752 RestartSpec {
753 workspace: id,
754 exec: exec.clone(),
755 argv: argv.clone(),
756 envp: envp.clone(),
757 backoff_ms: 200,
758 max_backoff_ms: 30_000,
759 restart_count: 0,
760 },
761 );
762 }
763 }
764 for d in &out.degradations {
765 warn!(?d, %id, "command incarnation degradation");
766 }
767 Ok(CommandSummary {
768 id: cmd_id,
769 label: cmd_label,
770 pid: pid.as_raw(),
771 })
772 }
773
774 /// Devuelve el tail del log capturado para `(workspace, command)`.
775 /// `stream` selecciona stdout/stderr/both.
776 pub async fn get_command_logs(
777 &self,
778 workspace: WorkspaceId,
779 command: Ulid,
780 tail_bytes: usize,
781 stream: LogStream,
782 ) -> Option<Vec<u8>> {
783 let g = self.inner.lock().await;
784 let ws = g.workspaces.get(&workspace)?;
785 let cmd = ws.commands.get(&command)?;
786 match stream {
787 LogStream::Stdout => cmd.stdout.as_ref().map(|lb| lb.tail(tail_bytes)),
788 LogStream::Stderr => cmd.stderr.as_ref().map(|lb| lb.tail(tail_bytes)),
789 LogStream::Both => {
790 let so = cmd.stdout.as_ref().map(|lb| lb.tail(tail_bytes)).unwrap_or_default();
791 let se = cmd.stderr.as_ref().map(|lb| lb.tail(tail_bytes)).unwrap_or_default();
792 let mut out = so;
793 out.extend_from_slice(&se);
794 Some(out)
795 }
796 }
797 }
798
799 /// Lista comandos de un workspace.
800 pub async fn list_commands(&self, workspace: WorkspaceId) -> Vec<CommandInfo> {
801 let g = self.inner.lock().await;
802 let Some(ws) = g.workspaces.get(&workspace) else { return Vec::new() };
803 let mut out: Vec<CommandInfo> = ws
804 .commands
805 .values()
806 .map(|c| CommandInfo {
807 id: c.id,
808 label: c.label.clone(),
809 pid: c.pid.as_raw(),
810 alive: c.alive,
811 exit_status: c.exit_status,
812 log_bytes: c.stdout.as_ref().map(|l| l.written_total()).unwrap_or(0)
813 + c.stderr.as_ref().map(|l| l.written_total()).unwrap_or(0),
814 })
815 .collect();
816 // Orden estable por ULID (temporal).
817 out.sort_by_key(|c| c.id);
818 out
819 }
820
821 /// Lanza todas las Cards de un Pipeline. Devuelve (label, pid) por nodo.
822 /// La conexión via flows queda librada al broker (cuando haya integración
823 /// completa con sidecar; v1 sólo lanza).
824 pub async fn run_pipeline(
825 &self,
826 spec: &PipelineSpec,
827 ) -> Result<Vec<(String, Pid)>, CoreError> {
828 spec.validate()?;
829 let workspace_label = {
830 let g = self.inner.lock().await;
831 let ws = g
832 .workspaces
833 .get(&spec.workspace)
834 .ok_or(CoreError::WorkspaceNotFound(spec.workspace))?;
835 ws.spec.label.clone()
836 };
837 let mut launched = Vec::new();
838 for (i, node) in spec.nodes.iter().enumerate() {
839 let card = node.to_card(i, &workspace_label)?;
840 let out = self.incarnator.incarnate(&card)?;
841 let mut g = self.inner.lock().await;
842 if let Some(ws) = g.workspaces.get_mut(&spec.workspace) {
843 ws.commands.insert(
844 card.id,
845 CommandState {
846 id: card.id,
847 label: node.label.clone(),
848 pid: out.pid,
849 alive: true,
850 exit_status: None,
851 stdout: None, // run_pipeline NO captura (conecta por pipes).
852 stderr: None,
853 pipeline_id: None,
854 },
855 );
856 }
857 launched.push((node.label.clone(), out.pid));
858 }
859 Ok(launched)
860 }
861
862 /// Cosecha hijos terminados (no-bloqueante). Llamar periódicamente desde
863 /// el daemon o ante SIGCHLD. Marca `alive=false` y guarda exit_status.
864 pub async fn reap_dead(self: &Arc<Self>) {
865 let mut to_restart: Vec<RestartSpec> = Vec::new();
866 let mut to_enforce_kill: Vec<WorkspaceId> = Vec::new();
867 {
868 let mut g = self.inner.lock().await;
869 for ws in g.workspaces.values_mut() {
870 for cmd in ws.commands.values_mut() {
871 if !cmd.alive {
872 continue;
873 }
874 match waitpid(cmd.pid, Some(WaitPidFlag::WNOHANG)) {
875 Ok(WaitStatus::Exited(_, code)) => {
876 cmd.alive = false;
877 cmd.exit_status = Some(code);
878 }
879 Ok(WaitStatus::Signaled(_, sig, _)) => {
880 cmd.alive = false;
881 cmd.exit_status = Some(128 + (sig as i32));
882 }
883 _ => {}
884 }
885 }
886 }
887 // Quota enforcement: chequear breach por workspace y aplicar policy.
888 // Lo hacemos dentro del mismo lock para tener una lectura
889 // consistente; el kill real va fuera del lock.
890 for (ws_id, ws) in g.workspaces.iter() {
891 let rl = &ws.spec.soma.rlimits;
892 let qe = &ws.spec.quota_enforce;
893 // Sólo aplicamos si hay al menos una action != None.
894 if qe.mem == shipote_card::QuotaAction::None
895 && qe.nproc == shipote_card::QuotaAction::None
896 {
897 continue;
898 }
899 // Medir RSS y nproc vivos sin pasar por workspace_stats
900 // (que tomaría el lock recursivo). Hacemos un read directo.
901 let alive: Vec<i32> = ws
902 .commands
903 .values()
904 .filter(|c| c.alive)
905 .map(|c| c.pid.as_raw())
906 .collect();
907 let nproc_alive = alive.len() as u32;
908 let mem_used: u64 = alive
909 .iter()
910 .filter_map(|pid| read_proc_rss(*pid))
911 .sum();
912
913 let mem_breach = matches!(rl.mem_bytes, Some(limit) if mem_used > limit);
914 let nproc_breach = matches!(rl.nproc, Some(limit) if nproc_alive > limit);
915
916 let mut kill_needed = false;
917 if mem_breach {
918 match qe.mem {
919 shipote_card::QuotaAction::Log => {
920 warn!(%ws_id, used = mem_used, limit = ?rl.mem_bytes, "quota breach: memory");
921 }
922 shipote_card::QuotaAction::Kill => {
923 warn!(%ws_id, used = mem_used, limit = ?rl.mem_bytes, "quota breach: KILLING");
924 kill_needed = true;
925 }
926 _ => {}
927 }
928 }
929 if nproc_breach {
930 match qe.nproc {
931 shipote_card::QuotaAction::Log => {
932 warn!(%ws_id, alive = nproc_alive, limit = ?rl.nproc, "quota breach: nproc");
933 }
934 shipote_card::QuotaAction::Kill => {
935 warn!(%ws_id, alive = nproc_alive, limit = ?rl.nproc, "quota breach: KILLING");
936 kill_needed = true;
937 }
938 _ => {}
939 }
940 }
941 if kill_needed {
942 to_enforce_kill.push(*ws_id);
943 }
944 }
945 // Pipeline supervisor: detectar pipelines cuyos comandos tienen
946 // failure. Marca para restart si tiene supervisor.
947 // Esto se hace cuando TODOS los comandos del pipeline están
948 // dead Y al menos uno tiene exit!=0 (sino podría disparar
949 // restart mientras otros comandos aún corren — incorrecto).
950 let supervisor_ids: Vec<Ulid> = g.pipeline_supervisors.keys().copied().collect();
951 for pipe_id in supervisor_ids {
952 // ¿Hay algún comando vivo de este pipeline?
953 let mut all_dead = true;
954 let mut any_failed = false;
955 for ws in g.workspaces.values() {
956 for cmd in ws.commands.values() {
957 if cmd.pipeline_id != Some(pipe_id) {
958 continue;
959 }
960 if cmd.alive {
961 all_dead = false;
962 } else if cmd.exit_status.map_or(false, |s| s != 0) {
963 any_failed = true;
964 }
965 }
966 }
967 if all_dead && any_failed {
968 // Push a queue si no estaba ya.
969 if !g.pending_pipeline_restarts.contains(&pipe_id) {
970 g.pending_pipeline_restarts.push(pipe_id);
971 }
972 }
973 }
974 // Detectar restart_specs cuyo command_id ya está dead con exit!=0.
975 let mut to_remove: Vec<Ulid> = Vec::new();
976 for (cmd_id, spec) in g.restart_specs.iter() {
977 let mut should_restart = false;
978 let mut should_drop = false;
979 'outer: for ws in g.workspaces.values() {
980 if let Some(cmd) = ws.commands.get(cmd_id) {
981 if !cmd.alive {
982 match cmd.exit_status {
983 Some(0) => should_drop = true,
984 Some(_) => should_restart = true,
985 None => {}
986 }
987 break 'outer;
988 }
989 }
990 }
991 if should_drop {
992 to_remove.push(*cmd_id);
993 } else if should_restart {
994 to_restart.push(spec.clone());
995 to_remove.push(*cmd_id);
996 }
997 }
998 for id in to_remove {
999 g.restart_specs.remove(&id);
1000 }
1001 }
1002 // Quota enforcement: kill workspaces fuera del lock.
1003 for ws_id in to_enforce_kill {
1004 let _ = self.stop_with_grace(ws_id, std::time::Duration::ZERO).await;
1005 }
1006 // Schedule restart fuera del lock.
1007 for mut spec in to_restart {
1008 let mgr = self.clone();
1009 let backoff = std::time::Duration::from_millis(spec.backoff_ms);
1010 // Subir el backoff para la PRÓXIMA falla, no esta.
1011 spec.backoff_ms = (spec.backoff_ms * 2).min(spec.max_backoff_ms);
1012 spec.restart_count += 1;
1013 let restart_n = spec.restart_count;
1014 tokio::spawn(async move {
1015 tokio::time::sleep(backoff).await;
1016 info!(
1017 backoff_ms = backoff.as_millis() as u64,
1018 restart = restart_n,
1019 "restarting failed command"
1020 );
1021 let workspace = spec.workspace;
1022 if let Err(e) = mgr
1023 .run_with_options(workspace, spec.exec.clone(), spec.argv.clone(), spec.envp.clone(), true)
1024 .await
1025 {
1026 warn!(?e, "restart failed to launch");
1027 return;
1028 }
1029 // Preservar backoff acumulado: localizar el nuevo command_id
1030 // (el más reciente vivo en el workspace) y sobreescribir.
1031 let new_cmd_id = {
1032 let g = mgr.inner.lock().await;
1033 g.workspaces.get(&workspace).and_then(|ws| {
1034 ws.commands
1035 .values()
1036 .filter(|c| c.alive)
1037 .max_by_key(|c| c.id)
1038 .map(|c| c.id)
1039 })
1040 };
1041 if let Some(new_id) = new_cmd_id {
1042 let mut g = mgr.inner.lock().await;
1043 if let Some(existing) = g.restart_specs.get_mut(&new_id) {
1044 existing.backoff_ms = spec.backoff_ms;
1045 existing.restart_count = spec.restart_count;
1046 }
1047 }
1048 });
1049 }
1050 }
1051}
1052
1053#[derive(Debug, Clone)]
1054pub struct WorkspaceSnapshot {
1055 pub id: WorkspaceId,
1056 pub label: String,
1057 pub commands: u32,
1058 pub uptime_ms: u64,
1059}
1060
1061fn short_ulid(u: &Ulid) -> String {
1062 let s = u.to_string();
1063 s[s.len() - 6..].to_string()
1064}
1065
1066#[cfg(test)]
1067mod tests {
1068 use super::*;
1069
1070 #[tokio::test]
1071 async fn ttl_auto_stops_workspace() {
1072 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1073 let spec = WorkspaceSpec {
1074 label: "ttl-test".into(),
1075 soma: Default::default(),
1076 permissions: Default::default(),
1077 ttl: Some(std::time::Duration::from_millis(120)),
1078 flow_dirs: vec![],
1079 on_exit: shipote_card::ExitPolicy::Reap,
1080 quota_enforce: Default::default(),
1081 };
1082 let (id, _) = mgr.create(spec).await.unwrap();
1083 assert_eq!(mgr.list().await.len(), 1);
1084 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
1085 assert_eq!(
1086 mgr.list().await.len(),
1087 0,
1088 "TTL expirado: workspace debe haber sido removido"
1089 );
1090 let _ = id;
1091 }
1092
1093 #[tokio::test]
1094 async fn create_and_list_workspace() {
1095 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1096 let spec = WorkspaceSpec {
1097 label: "test".into(),
1098 soma: Default::default(),
1099 permissions: Default::default(),
1100 ttl: None,
1101 flow_dirs: vec![],
1102 on_exit: shipote_card::ExitPolicy::Reap,
1103 quota_enforce: Default::default(),
1104 };
1105 let (id, _w) = mgr.create(spec).await.unwrap();
1106 let list = mgr.list().await;
1107 assert_eq!(list.len(), 1);
1108 assert_eq!(list[0].id, id);
1109 }
1110
1111 #[tokio::test]
1112 async fn run_captures_stdout_to_log() {
1113 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1114 let spec = WorkspaceSpec {
1115 label: "logs".into(),
1116 soma: Default::default(),
1117 permissions: Default::default(),
1118 ttl: None,
1119 flow_dirs: vec![],
1120 on_exit: shipote_card::ExitPolicy::Reap,
1121 quota_enforce: Default::default(),
1122 };
1123 let (id, _) = mgr.create(spec).await.unwrap();
1124 let summary = mgr
1125 .run(id, "/bin/echo".into(), vec!["captured-output".into()], vec![])
1126 .await
1127 .unwrap();
1128 for _ in 0..50 {
1129 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1130 mgr.reap_dead().await;
1131 let logs = mgr
1132 .get_command_logs(id, summary.id, 0, LogStream::Stdout)
1133 .await
1134 .unwrap_or_default();
1135 if !logs.is_empty() {
1136 let s = String::from_utf8_lossy(&logs);
1137 assert!(s.contains("captured-output"), "got: {s:?}");
1138 return;
1139 }
1140 }
1141 panic!("logs never captured");
1142 }
1143
1144 #[tokio::test]
1145 async fn run_captures_stderr_separately() {
1146 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1147 let spec = WorkspaceSpec {
1148 label: "stderr".into(),
1149 soma: Default::default(),
1150 permissions: Default::default(),
1151 ttl: None,
1152 flow_dirs: vec![],
1153 on_exit: shipote_card::ExitPolicy::Reap,
1154 quota_enforce: Default::default(),
1155 };
1156 let (id, _) = mgr.create(spec).await.unwrap();
1157 // sh -c "echo OUT; echo ERR >&2"
1158 let summary = mgr
1159 .run(
1160 id,
1161 "/bin/sh".into(),
1162 vec!["-c".into(), "echo OUT; echo ERR >&2".into()],
1163 vec![],
1164 )
1165 .await
1166 .unwrap();
1167 for _ in 0..50 {
1168 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1169 mgr.reap_dead().await;
1170 let so = mgr
1171 .get_command_logs(id, summary.id, 0, LogStream::Stdout)
1172 .await
1173 .unwrap_or_default();
1174 let se = mgr
1175 .get_command_logs(id, summary.id, 0, LogStream::Stderr)
1176 .await
1177 .unwrap_or_default();
1178 if !so.is_empty() && !se.is_empty() {
1179 let so_s = String::from_utf8_lossy(&so);
1180 let se_s = String::from_utf8_lossy(&se);
1181 assert!(so_s.contains("OUT"), "stdout: {so_s:?}");
1182 assert!(se_s.contains("ERR"), "stderr: {se_s:?}");
1183 assert!(!so_s.contains("ERR"), "stdout no debería tener ERR");
1184 assert!(!se_s.contains("OUT"), "stderr no debería tener OUT");
1185 return;
1186 }
1187 }
1188 panic!("logs never captured on both streams");
1189 }
1190
1191 #[tokio::test]
1192 async fn restart_on_failure_relaunches_failing_command() {
1193 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1194 let spec = WorkspaceSpec {
1195 label: "restart".into(),
1196 soma: Default::default(),
1197 permissions: Default::default(),
1198 ttl: None,
1199 flow_dirs: vec![],
1200 on_exit: shipote_card::ExitPolicy::Reap,
1201 quota_enforce: Default::default(),
1202 };
1203 let (id, _) = mgr.create(spec).await.unwrap();
1204 // /bin/false sale con exit=1. Con restart_on_failure=true debería
1205 // relanzarse al menos 1 vez (tras el backoff inicial de 200ms).
1206 let summary = mgr
1207 .run_with_options(id, "/bin/false".into(), vec![], vec![], true)
1208 .await
1209 .unwrap();
1210 let original_id = summary.id;
1211 // Esperamos ~500ms para que termine + reap + restart corra.
1212 for _ in 0..30 {
1213 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1214 mgr.reap_dead().await;
1215 let g = mgr.inner.lock().await;
1216 if let Some(ws) = g.workspaces.get(&id) {
1217 let new_cmds: Vec<_> = ws.commands.keys().filter(|k| **k != original_id).collect();
1218 if !new_cmds.is_empty() {
1219 // Hay un nuevo command_id → restart funcionó.
1220 return;
1221 }
1222 }
1223 }
1224 panic!("restart never launched a new command");
1225 }
1226
1227 #[tokio::test]
1228 async fn pipeline_supervisor_queues_restart_on_failure() {
1229 use shipote_card::{CommandRef, DiscernPolicy, PipelineSpec};
1230 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1231 let (ws_id, _) = mgr.create(WorkspaceSpec {
1232 label: "psup".into(),
1233 soma: Default::default(),
1234 permissions: Default::default(),
1235 ttl: None,
1236 flow_dirs: vec![],
1237 on_exit: shipote_card::ExitPolicy::Reap,
1238 quota_enforce: Default::default(),
1239 }).await.unwrap();
1240 let spec = PipelineSpec {
1241 label: "fail-pipeline".into(),
1242 workspace: ws_id,
1243 nodes: vec![CommandRef {
1244 label: "boom".into(),
1245 payload: brahman_card::Payload::Native {
1246 exec: "/bin/false".into(),
1247 argv: vec![],
1248 envp: vec![],
1249 },
1250 soma: Default::default(),
1251 flows: Default::default(),
1252 supervision: brahman_card::Supervision::OneShot,
1253 }],
1254 edges: vec![],
1255 discern: DiscernPolicy::default(),
1256 restart_on_failure: true,
1257 };
1258 let pipeline_id = ulid::Ulid::new();
1259 // Simulamos lo que haría el daemon: registramos un comando como
1260 // si fuera de pipeline. Usamos `register_pipeline_commands` con
1261 // un pid fake — pero como reaper hace waitpid, mejor lanzar de verdad.
1262 // Hack: usar /bin/false via run() y manualmente marcar pipeline_id.
1263 let summary = mgr.run(ws_id, "/bin/false".into(), vec![], vec![]).await.unwrap();
1264 // Marcar el comando con pipeline_id manualmente.
1265 {
1266 let mut g = mgr.inner.lock().await;
1267 if let Some(ws) = g.workspaces.get_mut(&ws_id) {
1268 if let Some(cmd) = ws.commands.get_mut(&summary.id) {
1269 cmd.pipeline_id = Some(pipeline_id);
1270 }
1271 }
1272 }
1273 mgr.register_pipeline_supervisor(pipeline_id, ws_id, spec, true).await;
1274 // Esperamos que reap detecte la falla y push a pending.
1275 for _ in 0..40 {
1276 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1277 mgr.reap_dead().await;
1278 let pending = mgr.take_pending_restarts().await;
1279 if !pending.is_empty() {
1280 assert_eq!(pending[0].spec.label, "fail-pipeline");
1281 return;
1282 }
1283 }
1284 panic!("supervisor never queued a restart");
1285 }
1286
1287 #[tokio::test]
1288 async fn quota_enforce_nproc_kill_terminates_commands() {
1289 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1290 let mut spec = WorkspaceSpec {
1291 label: "qenforce".into(),
1292 soma: Default::default(),
1293 permissions: Default::default(),
1294 ttl: None,
1295 flow_dirs: vec![],
1296 on_exit: shipote_card::ExitPolicy::Reap,
1297 quota_enforce: shipote_card::QuotaEnforcement {
1298 mem: shipote_card::QuotaAction::None,
1299 nproc: shipote_card::QuotaAction::Kill,
1300 },
1301 };
1302 spec.soma.rlimits.nproc = Some(1);
1303 let (id, _) = mgr.create(spec).await.unwrap();
1304 // Lanzo 2 procesos (cada uno sleep). nproc_limit=1 → breach inmediato.
1305 let _ = mgr.run(id, "/bin/sleep".into(), vec!["5".into()], vec![]).await.unwrap();
1306 let _ = mgr.run(id, "/bin/sleep".into(), vec!["5".into()], vec![]).await.unwrap();
1307 // Reaper detecta breach y mata workspace.
1308 for _ in 0..30 {
1309 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1310 mgr.reap_dead().await;
1311 let alive = mgr.list().await;
1312 if alive.is_empty() {
1313 return; // workspace removido por stop()
1314 }
1315 }
1316 panic!("quota enforce kill never triggered");
1317 }
1318
1319 #[tokio::test]
1320 async fn run_true_in_workspace() {
1321 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
1322 let spec = WorkspaceSpec {
1323 label: "exec".into(),
1324 soma: Default::default(),
1325 permissions: Default::default(),
1326 ttl: None,
1327 flow_dirs: vec![],
1328 on_exit: shipote_card::ExitPolicy::Reap,
1329 quota_enforce: Default::default(),
1330 };
1331 let (id, _) = mgr.create(spec).await.unwrap();
1332 let summary = mgr
1333 .run(id, "/bin/true".into(), vec![], vec![])
1334 .await
1335 .unwrap();
1336 assert!(summary.pid > 0);
1337 // Cosecha.
1338 std::thread::sleep(std::time::Duration::from_millis(100));
1339 mgr.reap_dead().await;
1340 }
1341}