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