⎇ c22d2480b93b7802be8e8ea5a0075b8175c4fedc
crates/modules/shipote/shipote-core/src/lib.rs
History
Who touched what
raw
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 logbuf;
15pub mod persist;
16pub mod pipeline;
17
18use brahman_card::{Card, Payload, Supervision};
19use ente_incarnate::{Incarnator, IncarnatorConfig};
20use nix::sys::signal::{kill, Signal};
21use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
22use nix::unistd::Pid;
23use shipote_card::{CommandRef, PipelineSpec, WorkspaceId, WorkspaceSpec};
24use std::collections::HashMap;
25use std::sync::Arc;
26use std::time::Instant;
27use thiserror::Error;
28use tokio::sync::Mutex;
29use tracing::{info, warn};
30use ulid::Ulid;
31
32#[derive(Debug, Error)]
33pub enum CoreError {
34 #[error("workspace {0} not found")]
35 WorkspaceNotFound(WorkspaceId),
36 #[error("compile: {0}")]
37 Compile(#[from] shipote_card::CompileError),
38 #[error("incarnate: {0}")]
39 Incarnate(#[from] ente_incarnate::IncarnateError),
40}
41
42#[derive(Debug)]
43pub struct WorkspaceState {
44 pub id: WorkspaceId,
45 pub spec: WorkspaceSpec,
46 pub root_card: Card,
47 pub commands: HashMap<Ulid, CommandState>,
48 pub started: Instant,
49}
50
51#[derive(Debug, Clone)]
52pub struct CommandState {
53 pub id: Ulid,
54 pub label: String,
55 pub pid: Pid,
56 pub alive: bool,
57 pub exit_status: Option<i32>,
58 /// Ring buffer compartido con la tokio task que drena stdout+stderr
59 /// del comando. `None` para comandos que no capturan output (futuro:
60 /// comandos con stdout=inherit).
61 pub logs: Option<logbuf::LogBuf>,
62}
63
64pub struct WorkspaceManager {
65 inner: Arc<Mutex<Inner>>,
66 incarnator: Arc<Incarnator>,
67}
68
69struct Inner {
70 workspaces: HashMap<WorkspaceId, WorkspaceState>,
71 /// Definiciones nombradas de pipelines persistidas. NO es lo mismo
72 /// que "pipelines vivos" — son specs guardados para reusar con
73 /// `run-saved`. Sobreviven restart vía snapshot.
74 saved_pipelines: HashMap<String, PipelineSpec>,
75}
76
77#[derive(Debug, Clone)]
78pub struct CommandSummary {
79 pub id: Ulid,
80 pub label: String,
81 pub pid: i32,
82}
83
84#[derive(Debug, Clone)]
85pub struct CommandInfo {
86 pub id: Ulid,
87 pub label: String,
88 pub pid: i32,
89 pub alive: bool,
90 pub exit_status: Option<i32>,
91 pub log_bytes: u64,
92}
93
94fn spawn_log_drainer(read_fd: std::os::fd::RawFd, logs: logbuf::LogBuf) {
95 // Marcar non-blocking + envolver en AsyncFd; igual patrón que el tap.
96 // SAFETY: F_SETFL sobre fd válido.
97 unsafe {
98 let flags = libc::fcntl(read_fd, libc::F_GETFL, 0);
99 if flags >= 0 {
100 libc::fcntl(read_fd, libc::F_SETFL, flags | libc::O_NONBLOCK);
101 }
102 }
103 tokio::spawn(async move {
104 // SAFETY: ownership del fd transferido al drainer task.
105 let owned = unsafe { std::os::fd::OwnedFd::from_raw_fd_compat(read_fd) };
106 let afd = match tokio::io::unix::AsyncFd::with_interest(owned, tokio::io::Interest::READABLE) {
107 Ok(a) => a,
108 Err(e) => {
109 tracing::warn!(?e, "log drainer AsyncFd failed");
110 return;
111 }
112 };
113 let mut buf = [0u8; 4096];
114 loop {
115 let mut guard = match afd.readable().await {
116 Ok(g) => g,
117 Err(_) => break,
118 };
119 use std::os::fd::AsRawFd;
120 let fd = afd.as_raw_fd();
121 // SAFETY: read sobre fd válido.
122 let r = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
123 if r > 0 {
124 logs.append(&buf[..r as usize]);
125 continue;
126 }
127 if r == 0 {
128 break; // EOF
129 }
130 let err = std::io::Error::last_os_error();
131 if err.kind() == std::io::ErrorKind::WouldBlock {
132 guard.clear_ready();
133 continue;
134 }
135 tracing::warn!(?err, "log drainer read err");
136 break;
137 }
138 });
139}
140
141trait OwnedFdFromRawCompat: Sized {
142 unsafe fn from_raw_fd_compat(fd: std::os::fd::RawFd) -> Self;
143}
144
145impl OwnedFdFromRawCompat for std::os::fd::OwnedFd {
146 unsafe fn from_raw_fd_compat(fd: std::os::fd::RawFd) -> Self {
147 use std::os::fd::FromRawFd;
148 // SAFETY: el caller transfiere ownership de fd a OwnedFd.
149 unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }
150 }
151}
152
153impl WorkspaceManager {
154 pub fn new(cfg: IncarnatorConfig) -> Self {
155 Self {
156 inner: Arc::new(Mutex::new(Inner {
157 workspaces: HashMap::new(),
158 saved_pipelines: HashMap::new(),
159 })),
160 incarnator: Arc::new(Incarnator::new(cfg)),
161 }
162 }
163
164 pub fn incarnator(&self) -> &Incarnator {
165 &self.incarnator
166 }
167
168 /// Handle Arc-clonable del Incarnator, para que el pipeline lo pueda
169 /// usar fuera del manager.
170 pub fn incarnator_handle(&self) -> Arc<Incarnator> {
171 self.incarnator.clone()
172 }
173
174 // -----------------------------------------------------------------
175 // Saved pipelines (definiciones nombradas, no runs)
176 // -----------------------------------------------------------------
177
178 /// Guarda (o reemplaza) un PipelineSpec bajo `name`.
179 pub async fn save_pipeline(&self, name: String, spec: PipelineSpec) {
180 self.inner.lock().await.saved_pipelines.insert(name, spec);
181 }
182
183 /// Devuelve los nombres de los pipelines guardados.
184 pub async fn list_saved_pipelines(&self) -> Vec<String> {
185 let g = self.inner.lock().await;
186 let mut v: Vec<String> = g.saved_pipelines.keys().cloned().collect();
187 v.sort();
188 v
189 }
190
191 /// Recupera el PipelineSpec guardado bajo `name`.
192 pub async fn get_saved_pipeline(&self, name: &str) -> Option<PipelineSpec> {
193 self.inner.lock().await.saved_pipelines.get(name).cloned()
194 }
195
196 /// Elimina un saved pipeline.
197 pub async fn drop_saved_pipeline(&self, name: &str) -> bool {
198 self.inner.lock().await.saved_pipelines.remove(name).is_some()
199 }
200
201 /// Label del workspace, si existe.
202 pub async fn workspace_label(&self, id: WorkspaceId) -> Option<String> {
203 self.inner
204 .lock()
205 .await
206 .workspaces
207 .get(&id)
208 .map(|w| w.spec.label.clone())
209 }
210
211 pub async fn create(
212 self: &Arc<Self>,
213 spec: WorkspaceSpec,
214 ) -> Result<(WorkspaceId, Vec<String>), CoreError> {
215 self.create_with_id(WorkspaceId::new(), spec).await
216 }
217
218 /// Variante que acepta el ID. Útil para restore_snapshot: preserva
219 /// ULIDs entre restarts, así clients que tracking workspace_id no se
220 /// rompen.
221 pub async fn create_with_id(
222 self: &Arc<Self>,
223 id: WorkspaceId,
224 spec: WorkspaceSpec,
225 ) -> Result<(WorkspaceId, Vec<String>), CoreError> {
226 let card = spec.to_card(id)?;
227 let warnings = self.incarnator.dry_run(&card).warnings;
228 let ttl = spec.ttl;
229 let state = WorkspaceState {
230 id,
231 spec,
232 root_card: card,
233 commands: HashMap::new(),
234 started: Instant::now(),
235 };
236 self.inner.lock().await.workspaces.insert(id, state);
237 info!(%id, ?ttl, "workspace created");
238
239 // Si tiene TTL, programar auto-stop. El task captura un weak ref
240 // al manager para no impedir que se dropée si el daemon termina.
241 if let Some(duration) = ttl {
242 let mgr_weak = Arc::downgrade(self);
243 tokio::spawn(async move {
244 tokio::time::sleep(duration).await;
245 if let Some(mgr) = mgr_weak.upgrade() {
246 let exists = mgr.inner.lock().await.workspaces.contains_key(&id);
247 if exists {
248 info!(%id, "workspace TTL expired — auto-stop");
249 let _ = mgr.stop(id).await;
250 }
251 }
252 });
253 }
254
255 Ok((id, warnings))
256 }
257
258 pub async fn list(&self) -> Vec<WorkspaceSnapshot> {
259 let g = self.inner.lock().await;
260 g.workspaces
261 .values()
262 .map(|w| WorkspaceSnapshot {
263 id: w.id,
264 label: w.spec.label.clone(),
265 commands: w.commands.len() as u32,
266 uptime_ms: w.started.elapsed().as_millis() as u64,
267 })
268 .collect()
269 }
270
271 pub async fn stop(&self, id: WorkspaceId) -> Result<u32, CoreError> {
272 let mut g = self.inner.lock().await;
273 let ws = g.workspaces.remove(&id).ok_or(CoreError::WorkspaceNotFound(id))?;
274 let mut reaped = 0u32;
275 for (_cid, cmd) in ws.commands {
276 if cmd.alive {
277 let _ = kill(cmd.pid, Signal::SIGTERM);
278 // Cosecha sin bloquear infinito: WNOHANG en loop con un par de intentos.
279 for _ in 0..50 {
280 match waitpid(cmd.pid, Some(WaitPidFlag::WNOHANG)) {
281 Ok(WaitStatus::StillAlive) => {
282 std::thread::sleep(std::time::Duration::from_millis(20));
283 }
284 Ok(_) => {
285 reaped += 1;
286 break;
287 }
288 Err(_) => break,
289 }
290 }
291 // Último recurso: SIGKILL.
292 let _ = kill(cmd.pid, Signal::SIGKILL);
293 let _ = waitpid(cmd.pid, None);
294 }
295 }
296 info!(%id, reaped, "workspace stopped");
297 Ok(reaped)
298 }
299
300 /// Ejecuta un comando one-shot dentro de un workspace existente.
301 /// Captura stdout+stderr en un ring buffer accesible vía
302 /// [`get_command_logs`](Self::get_command_logs).
303 pub async fn run(
304 &self,
305 id: WorkspaceId,
306 exec: String,
307 argv: Vec<String>,
308 envp: Vec<(String, String)>,
309 ) -> Result<CommandSummary, CoreError> {
310 let workspace_label = {
311 let g = self.inner.lock().await;
312 let ws = g.workspaces.get(&id).ok_or(CoreError::WorkspaceNotFound(id))?;
313 ws.spec.label.clone()
314 };
315 let cmd_ref = CommandRef {
316 label: format!("run-{}", short_ulid(&Ulid::new())),
317 payload: Payload::Native { exec, argv, envp },
318 soma: Default::default(),
319 flows: Default::default(),
320 supervision: Supervision::OneShot,
321 };
322 let card = cmd_ref.to_card(0, &workspace_label)?;
323
324 // Pipe para capturar stdout. O_CLOEXEC para que hijos del hijo
325 // no hereden la copia. v1: stderr=inherit (simplicidad; tail útil
326 // para stdout solo). Futuro: stderr separado en el ring.
327 let (capture_r, capture_w) =
328 nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC).map_err(|e| {
329 CoreError::Incarnate(ente_incarnate::IncarnateError::Pipe(e))
330 })?;
331 use std::os::fd::IntoRawFd;
332 let capture_r_fd = capture_r.into_raw_fd();
333 let capture_w_fd = capture_w.into_raw_fd();
334
335 let logs = logbuf::LogBuf::new();
336
337 let stdio = ente_incarnate::ChildStdio {
338 stdin_fd: None,
339 stdout_fd: Some(capture_w_fd),
340 stderr_fd: None,
341 };
342 let out = self.incarnator.incarnate_with(&card, stdio)?;
343 let cmd_id = card.id;
344 let cmd_label = cmd_ref.label.clone();
345 let pid = out.pid;
346
347 // Drainer: tokio task que lee capture_r_fd y appendea al ring.
348 spawn_log_drainer(capture_r_fd, logs.clone());
349
350 let mut g = self.inner.lock().await;
351 if let Some(ws) = g.workspaces.get_mut(&id) {
352 ws.commands.insert(
353 cmd_id,
354 CommandState {
355 id: cmd_id,
356 label: cmd_label.clone(),
357 pid,
358 alive: true,
359 exit_status: None,
360 logs: Some(logs),
361 },
362 );
363 }
364 for d in &out.degradations {
365 warn!(?d, %id, "command incarnation degradation");
366 }
367 Ok(CommandSummary {
368 id: cmd_id,
369 label: cmd_label,
370 pid: pid.as_raw(),
371 })
372 }
373
374 /// Devuelve el tail del log capturado para `(workspace, command)`.
375 pub async fn get_command_logs(
376 &self,
377 workspace: WorkspaceId,
378 command: Ulid,
379 tail_bytes: usize,
380 ) -> Option<Vec<u8>> {
381 let g = self.inner.lock().await;
382 let ws = g.workspaces.get(&workspace)?;
383 let cmd = ws.commands.get(&command)?;
384 cmd.logs.as_ref().map(|lb| lb.tail(tail_bytes))
385 }
386
387 /// Lista comandos de un workspace.
388 pub async fn list_commands(&self, workspace: WorkspaceId) -> Vec<CommandInfo> {
389 let g = self.inner.lock().await;
390 let Some(ws) = g.workspaces.get(&workspace) else { return Vec::new() };
391 let mut out: Vec<CommandInfo> = ws
392 .commands
393 .values()
394 .map(|c| CommandInfo {
395 id: c.id,
396 label: c.label.clone(),
397 pid: c.pid.as_raw(),
398 alive: c.alive,
399 exit_status: c.exit_status,
400 log_bytes: c.logs.as_ref().map(|l| l.written_total()).unwrap_or(0),
401 })
402 .collect();
403 // Orden estable por ULID (temporal).
404 out.sort_by_key(|c| c.id);
405 out
406 }
407
408 /// Lanza todas las Cards de un Pipeline. Devuelve (label, pid) por nodo.
409 /// La conexión via flows queda librada al broker (cuando haya integración
410 /// completa con sidecar; v1 sólo lanza).
411 pub async fn run_pipeline(
412 &self,
413 spec: &PipelineSpec,
414 ) -> Result<Vec<(String, Pid)>, CoreError> {
415 spec.validate()?;
416 let workspace_label = {
417 let g = self.inner.lock().await;
418 let ws = g
419 .workspaces
420 .get(&spec.workspace)
421 .ok_or(CoreError::WorkspaceNotFound(spec.workspace))?;
422 ws.spec.label.clone()
423 };
424 let mut launched = Vec::new();
425 for (i, node) in spec.nodes.iter().enumerate() {
426 let card = node.to_card(i, &workspace_label)?;
427 let out = self.incarnator.incarnate(&card)?;
428 let mut g = self.inner.lock().await;
429 if let Some(ws) = g.workspaces.get_mut(&spec.workspace) {
430 ws.commands.insert(
431 card.id,
432 CommandState {
433 id: card.id,
434 label: node.label.clone(),
435 pid: out.pid,
436 alive: true,
437 exit_status: None,
438 logs: None, // run_pipeline NO captura logs (los conecta por pipes).
439 },
440 );
441 }
442 launched.push((node.label.clone(), out.pid));
443 }
444 Ok(launched)
445 }
446
447 /// Cosecha hijos terminados (no-bloqueante). Llamar periódicamente desde
448 /// el daemon o ante SIGCHLD. Marca `alive=false` y guarda exit_status.
449 pub async fn reap_dead(&self) {
450 let mut g = self.inner.lock().await;
451 for ws in g.workspaces.values_mut() {
452 for cmd in ws.commands.values_mut() {
453 if !cmd.alive {
454 continue;
455 }
456 match waitpid(cmd.pid, Some(WaitPidFlag::WNOHANG)) {
457 Ok(WaitStatus::Exited(_, code)) => {
458 cmd.alive = false;
459 cmd.exit_status = Some(code);
460 }
461 Ok(WaitStatus::Signaled(_, sig, _)) => {
462 cmd.alive = false;
463 cmd.exit_status = Some(128 + (sig as i32));
464 }
465 _ => {}
466 }
467 }
468 }
469 }
470}
471
472#[derive(Debug, Clone)]
473pub struct WorkspaceSnapshot {
474 pub id: WorkspaceId,
475 pub label: String,
476 pub commands: u32,
477 pub uptime_ms: u64,
478}
479
480fn short_ulid(u: &Ulid) -> String {
481 let s = u.to_string();
482 s[s.len() - 6..].to_string()
483}
484
485#[cfg(test)]
486mod tests {
487 use super::*;
488
489 #[tokio::test]
490 async fn ttl_auto_stops_workspace() {
491 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
492 let spec = WorkspaceSpec {
493 label: "ttl-test".into(),
494 soma: Default::default(),
495 permissions: Default::default(),
496 ttl: Some(std::time::Duration::from_millis(120)),
497 flow_dirs: vec![],
498 on_exit: shipote_card::ExitPolicy::Reap,
499 };
500 let (id, _) = mgr.create(spec).await.unwrap();
501 assert_eq!(mgr.list().await.len(), 1);
502 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
503 assert_eq!(
504 mgr.list().await.len(),
505 0,
506 "TTL expirado: workspace debe haber sido removido"
507 );
508 let _ = id;
509 }
510
511 #[tokio::test]
512 async fn create_and_list_workspace() {
513 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
514 let spec = WorkspaceSpec {
515 label: "test".into(),
516 soma: Default::default(),
517 permissions: Default::default(),
518 ttl: None,
519 flow_dirs: vec![],
520 on_exit: shipote_card::ExitPolicy::Reap,
521 };
522 let (id, _w) = mgr.create(spec).await.unwrap();
523 let list = mgr.list().await;
524 assert_eq!(list.len(), 1);
525 assert_eq!(list[0].id, id);
526 }
527
528 #[tokio::test]
529 async fn run_captures_stdout_to_log() {
530 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
531 let spec = WorkspaceSpec {
532 label: "logs".into(),
533 soma: Default::default(),
534 permissions: Default::default(),
535 ttl: None,
536 flow_dirs: vec![],
537 on_exit: shipote_card::ExitPolicy::Reap,
538 };
539 let (id, _) = mgr.create(spec).await.unwrap();
540 let summary = mgr
541 .run(
542 id,
543 "/bin/echo".into(),
544 vec!["captured-output".into()],
545 vec![],
546 )
547 .await
548 .unwrap();
549 // Esperamos a que el comando termine y el drainer drene.
550 for _ in 0..50 {
551 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
552 mgr.reap_dead().await;
553 let logs = mgr.get_command_logs(id, summary.id, 0).await.unwrap_or_default();
554 if !logs.is_empty() {
555 let s = String::from_utf8_lossy(&logs);
556 assert!(s.contains("captured-output"), "got: {s:?}");
557 return;
558 }
559 }
560 panic!("logs never captured");
561 }
562
563 #[tokio::test]
564 async fn run_true_in_workspace() {
565 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
566 let spec = WorkspaceSpec {
567 label: "exec".into(),
568 soma: Default::default(),
569 permissions: Default::default(),
570 ttl: None,
571 flow_dirs: vec![],
572 on_exit: shipote_card::ExitPolicy::Reap,
573 };
574 let (id, _) = mgr.create(spec).await.unwrap();
575 let summary = mgr
576 .run(id, "/bin/true".into(), vec![], vec![])
577 .await
578 .unwrap();
579 assert!(summary.pid > 0);
580 // Cosecha.
581 std::thread::sleep(std::time::Duration::from_millis(100));
582 mgr.reap_dead().await;
583 }
584}