⎇ 4c9d1b4c1dbc62cf7d9cfde211524eb06ba21144
crates/modules/shipote/shipote-card/src/lib.rs
Historia
Quién tocó qué
crudo
feat(shipote): quota enforce + cgroup memory.max + pipeline restart (fase L)
1//! `shipote-card` — tipos del runtime shipote.
2//!
3//! Tres entidades nuevas encima del `brahman-card::Card`:
4//!
5//! - [`WorkspaceSpec`] — espacio aislado raíz con su propio `SomaSpec`.
6//! - [`CommandRef`] — un comando dentro de un workspace.
7//! - [`PipelineSpec`] — DAG de `CommandRef` conectados por `FlowEdge`.
8//!
9//! Cada `WorkspaceSpec`/`CommandRef` se **compila** a una o varias
10//! [`brahman_card::Card`] que el daemon entrega al [`Incarnator`] de
11//! `ente-incarnate`. Esto preserva el contrato canónico del fractal.
12
13#![forbid(unsafe_code)]
14
15use brahman_card::{Card, Payload, Permissions, SomaSpec, Supervision};
16use serde::{Deserialize, Serialize};
17use std::time::Duration;
18use thiserror::Error;
19use ulid::Ulid;
20
21// =====================================================================
22// Identidades
23// =====================================================================
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
26pub struct WorkspaceId(pub Ulid);
27
28impl WorkspaceId {
29 pub fn new() -> Self {
30 Self(Ulid::new())
31 }
32}
33
34impl Default for WorkspaceId {
35 fn default() -> Self {
36 Self::new()
37 }
38}
39
40impl std::fmt::Display for WorkspaceId {
41 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42 self.0.fmt(f)
43 }
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
47pub struct PipelineId(pub Ulid);
48
49impl PipelineId {
50 pub fn new() -> Self {
51 Self(Ulid::new())
52 }
53}
54
55impl Default for PipelineId {
56 fn default() -> Self {
57 Self::new()
58 }
59}
60
61impl std::fmt::Display for PipelineId {
62 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63 self.0.fmt(f)
64 }
65}
66
67// =====================================================================
68// Workspace
69// =====================================================================
70
71/// Espacio aislado de shipote. Es la raíz de aislamiento — cualquier comando
72/// que corre dentro hereda restricciones y no puede aflojarlas.
73#[derive(Debug, Clone, Serialize, Deserialize)]
74pub struct WorkspaceSpec {
75 pub label: String,
76
77 /// Aislamiento del workspace mismo (cuando se materializa como Card raíz).
78 #[serde(default)]
79 pub soma: SomaSpec,
80
81 /// Permisos máximos para hijas. Hijas pueden bajar pero no subir.
82 #[serde(default)]
83 pub permissions: Permissions,
84
85 /// `None` = vive hasta `stop`. `Some(d)` = el daemon lo termina tras d.
86 #[serde(default, with = "opt_duration_millis")]
87 pub ttl: Option<Duration>,
88
89 /// Slots de flow pre-declarados. Limitan qué consumidores externos al
90 /// workspace pueden empatar contra los productores internos.
91 #[serde(default)]
92 pub flow_dirs: Vec<FlowSlot>,
93
94 /// Política al terminar el workspace.
95 #[serde(default)]
96 pub on_exit: ExitPolicy,
97
98 /// Política de enforcement automático cuando un recurso excede su
99 /// rlimit declarado en `soma.rlimits`. Default = sólo accounting
100 /// (None) — el quota report sigue funcionando, pero no hay kill.
101 #[serde(default)]
102 pub quota_enforce: QuotaEnforcement,
103}
104
105/// Acción cuando un recurso excede su límite. Aplica por recurso (mem,
106/// nproc, ...).
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
108#[serde(rename_all = "lowercase")]
109pub enum QuotaAction {
110 /// Sólo accounting: la breach aparece en `workspace_quota`.
111 #[default]
112 None,
113 /// Loguear la breach (info-level del daemon).
114 Log,
115 /// Matar todos los comandos vivos del workspace (SIGKILL, sin grace).
116 Kill,
117}
118
119#[derive(Debug, Clone, Default, Serialize, Deserialize)]
120pub struct QuotaEnforcement {
121 #[serde(default)]
122 pub mem: QuotaAction,
123 #[serde(default)]
124 pub nproc: QuotaAction,
125}
126
127#[derive(Debug, Clone, Serialize, Deserialize)]
128pub struct FlowSlot {
129 pub name: String,
130 pub direction: FlowDirection,
131 /// Si `Workspace`, sólo otros nodos del mismo workspace pueden empatar.
132 /// Si `Public`, el broker global puede emparejar.
133 #[serde(default)]
134 pub scope: FlowScope,
135}
136
137#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
138#[serde(rename_all = "lowercase")]
139pub enum FlowDirection {
140 Input,
141 Output,
142}
143
144#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
145#[serde(rename_all = "lowercase")]
146pub enum FlowScope {
147 #[default]
148 Workspace,
149 Public,
150}
151
152#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
153#[serde(rename_all = "kebab-case")]
154pub enum ExitPolicy {
155 /// Reapear procesos hijos y descartar estado.
156 #[default]
157 Reap,
158 /// Mantener el workspace en `stopped` para inspección.
159 Keep,
160 /// Tomar snapshot del estado (para restart posterior).
161 Snapshot,
162}
163
164mod opt_duration_millis {
165 use serde::{Deserialize, Deserializer, Serialize, Serializer};
166 use std::time::Duration;
167
168 pub fn serialize<S: Serializer>(d: &Option<Duration>, s: S) -> Result<S::Ok, S::Error> {
169 d.map(|x| x.as_millis() as u64).serialize(s)
170 }
171
172 pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Option<Duration>, D::Error> {
173 let v: Option<u64> = Option::deserialize(d)?;
174 Ok(v.map(Duration::from_millis))
175 }
176}
177
178// =====================================================================
179// CommandRef
180// =====================================================================
181
182/// Un comando que vive dentro de un workspace. Se compila a una `Card` con
183/// `pin_to` apuntando al workspace padre (label) y su `SomaSpec`
184/// intersectado con el del workspace.
185#[derive(Debug, Clone, Serialize, Deserialize)]
186pub struct CommandRef {
187 pub label: String,
188 pub payload: Payload,
189
190 /// SomaSpec del comando. El compilador lo intersecta con el del workspace.
191 #[serde(default)]
192 pub soma: SomaSpec,
193
194 /// Inputs/outputs tipados (mismos `Flow` de brahman-card).
195 #[serde(default)]
196 pub flows: brahman_card::Flows,
197
198 /// Política de supervisión. Default `OneShot` (un comando se ejecuta y muere).
199 #[serde(default = "default_oneshot")]
200 pub supervision: Supervision,
201}
202
203fn default_oneshot() -> Supervision {
204 Supervision::OneShot
205}
206
207// =====================================================================
208// Pipeline
209// =====================================================================
210
211#[derive(Debug, Clone, Serialize, Deserialize)]
212pub struct PipelineSpec {
213 pub label: String,
214 pub workspace: WorkspaceId,
215 pub nodes: Vec<CommandRef>,
216 #[serde(default)]
217 pub edges: Vec<FlowEdge>,
218 #[serde(default)]
219 pub discern: DiscernPolicy,
220 /// Si `true` y cualquier comando del pipeline termina con exit!=0,
221 /// el daemon relaunch el pipeline ENTERO (stop + nuevo run_pipeline).
222 /// Útil para pipelines de procesamiento continuo.
223 #[serde(default)]
224 pub restart_on_failure: bool,
225}
226
227#[derive(Debug, Clone, Serialize, Deserialize)]
228pub struct FlowEdge {
229 /// Índice en `PipelineSpec.nodes` del productor.
230 pub from: usize,
231 /// Nombre del Flow output del productor.
232 pub from_output: String,
233 /// Índice en `PipelineSpec.nodes` del consumidor.
234 pub to: usize,
235 /// Nombre del Flow input del consumidor.
236 pub to_input: String,
237}
238
239#[derive(Debug, Clone, Serialize, Deserialize)]
240pub struct DiscernPolicy {
241 /// Bytes a samplear por flow para el discernidor. Default 4 KiB.
242 #[serde(default = "default_sample_bytes")]
243 pub sample_bytes: usize,
244 /// Si `true`, enriquece la Card del producer con el TypeRef detectado.
245 #[serde(default = "default_true")]
246 pub enrich_producer: bool,
247 /// Chunks que el FlowChannel guarda en replay buffer para subscribers
248 /// tarde. Default 32. Subir si los productores escriben en ráfagas y
249 /// querés que los consumidores tardíos vean toda la salida.
250 #[serde(default = "default_replay_chunks")]
251 pub replay_chunks: usize,
252 /// Tope adicional por **bytes** acumulados en el replay buffer. Lo
253 /// que se exceda primero (chunks o bytes) drop-ea el chunk más viejo.
254 /// `0` = sin tope por bytes (sólo aplica `replay_chunks`). Útil para
255 /// productores con chunks de tamaño variable.
256 #[serde(default)]
257 pub replay_bytes: usize,
258}
259
260impl Default for DiscernPolicy {
261 fn default() -> Self {
262 Self {
263 sample_bytes: default_sample_bytes(),
264 enrich_producer: default_true(),
265 replay_chunks: default_replay_chunks(),
266 replay_bytes: 0,
267 }
268 }
269}
270
271fn default_sample_bytes() -> usize {
272 4096
273}
274fn default_true() -> bool {
275 true
276}
277fn default_replay_chunks() -> usize {
278 32
279}
280
281// =====================================================================
282// Compilación a Card
283// =====================================================================
284
285#[derive(Debug, Error)]
286pub enum CompileError {
287 #[error("workspace label vacío")]
288 EmptyWorkspaceLabel,
289 #[error("comando con label vacío en posición {0}")]
290 EmptyCommandLabel(usize),
291 #[error("edge fuera de rango: from={from}, to={to}, nodes={nodes}")]
292 EdgeOutOfBounds { from: usize, to: usize, nodes: usize },
293}
294
295impl WorkspaceSpec {
296 /// Compila el WorkspaceSpec a una Card raíz que el Incarnator puede
297 /// encarnar. Usa `Payload::Virtual` (el workspace no es un proceso por
298 /// sí solo; sólo aloja hijos).
299 pub fn to_card(&self, id: WorkspaceId) -> Result<Card, CompileError> {
300 if self.label.trim().is_empty() {
301 return Err(CompileError::EmptyWorkspaceLabel);
302 }
303 let mut c = Card::new(format!("shipote.workspace.{}", self.label));
304 c.id = id.0;
305 c.soma = self.soma.clone();
306 c.permissions = self.permissions.clone();
307 c.payload = Payload::Virtual;
308 c.supervision = Supervision::OneShot;
309 Ok(c)
310 }
311}
312
313impl CommandRef {
314 /// Compila un CommandRef a Card hija de un workspace. La Card resultante
315 /// referencia al workspace por label en `pin_to` de cada Flow.
316 pub fn to_card(&self, idx: usize, workspace_label: &str) -> Result<Card, CompileError> {
317 if self.label.trim().is_empty() {
318 return Err(CompileError::EmptyCommandLabel(idx));
319 }
320 let mut c = Card::new(format!("shipote.cmd.{}.{}", workspace_label, self.label));
321 c.payload = self.payload.clone();
322 c.soma = intersect_soma(&self.soma, /*workspace*/ &SomaSpec::default());
323 c.supervision = self.supervision.clone();
324 c.flow = self.flows.clone();
325 // pin_to del workspace en cada Flow input/output → el broker prefiere
326 // resolver dentro del mismo workspace cuando hay candidatos múltiples.
327 let pin = format!("shipote.workspace.{}", workspace_label);
328 for f in c.flow.input.iter_mut().chain(c.flow.output.iter_mut()) {
329 if f.pin_to.is_none() {
330 f.pin_to = Some(pin.clone());
331 }
332 }
333 Ok(c)
334 }
335}
336
337/// Intersección conservadora: si el workspace pidió aislamiento, la hija
338/// también lo tiene (no puede aflojar). Si la hija pidió aislamiento extra,
339/// se respeta.
340fn intersect_soma(child: &SomaSpec, ws: &SomaSpec) -> SomaSpec {
341 let mut out = child.clone();
342 out.namespaces.mount |= ws.namespaces.mount;
343 out.namespaces.pid |= ws.namespaces.pid;
344 out.namespaces.net |= ws.namespaces.net;
345 out.namespaces.uts |= ws.namespaces.uts;
346 out.namespaces.ipc |= ws.namespaces.ipc;
347 out.namespaces.user |= ws.namespaces.user;
348 out.namespaces.cgroup |= ws.namespaces.cgroup;
349 // rlimits: el menor (más restrictivo) gana.
350 out.rlimits.mem_bytes = min_opt(out.rlimits.mem_bytes, ws.rlimits.mem_bytes);
351 out.rlimits.nproc = min_opt(out.rlimits.nproc, ws.rlimits.nproc);
352 out.rlimits.nofile = min_opt(out.rlimits.nofile, ws.rlimits.nofile);
353 out
354}
355
356fn min_opt<T: Ord + Copy>(a: Option<T>, b: Option<T>) -> Option<T> {
357 match (a, b) {
358 (Some(x), Some(y)) => Some(x.min(y)),
359 (Some(x), None) | (None, Some(x)) => Some(x),
360 (None, None) => None,
361 }
362}
363
364impl PipelineSpec {
365 pub fn validate(&self) -> Result<(), CompileError> {
366 let n = self.nodes.len();
367 for (i, c) in self.nodes.iter().enumerate() {
368 if c.label.trim().is_empty() {
369 return Err(CompileError::EmptyCommandLabel(i));
370 }
371 }
372 for e in &self.edges {
373 if e.from >= n || e.to >= n {
374 return Err(CompileError::EdgeOutOfBounds {
375 from: e.from,
376 to: e.to,
377 nodes: n,
378 });
379 }
380 }
381 Ok(())
382 }
383}
384
385// =====================================================================
386// I/O conveniencia (TOML + JSON)
387// =====================================================================
388
389#[derive(Debug, Error)]
390pub enum LoadError {
391 #[error("io: {0}")]
392 Io(#[from] std::io::Error),
393 #[error("toml: {0}")]
394 Toml(#[from] toml::de::Error),
395 #[error("json: {0}")]
396 Json(#[from] serde_json::Error),
397 #[error("formato desconocido (esperado .toml o .json)")]
398 UnknownFormat,
399}
400
401pub fn load_workspace_spec(path: &std::path::Path) -> Result<WorkspaceSpec, LoadError> {
402 let raw = std::fs::read_to_string(path)?;
403 match path.extension().and_then(|s| s.to_str()) {
404 Some("toml") => Ok(toml::from_str(&raw)?),
405 Some("json") => Ok(serde_json::from_str(&raw)?),
406 _ => Err(LoadError::UnknownFormat),
407 }
408}
409
410pub fn load_pipeline_spec(path: &std::path::Path) -> Result<PipelineSpec, LoadError> {
411 let raw = std::fs::read_to_string(path)?;
412 match path.extension().and_then(|s| s.to_str()) {
413 Some("toml") => Ok(toml::from_str(&raw)?),
414 Some("json") => Ok(serde_json::from_str(&raw)?),
415 _ => Err(LoadError::UnknownFormat),
416 }
417}
418
419/// Sustituye `${KEY}` en todos los strings del spec por el valor de
420/// `vars["KEY"]`. Variables sin match quedan intactas (no se borra el
421/// placeholder — útil para detectar olvidos).
422///
423/// Walk recursivo sobre la representación JSON intermedia para cubrir
424/// labels, argv, envp, paths y cualquier String del schema.
425pub fn substitute_vars(
426 spec: &PipelineSpec,
427 vars: &std::collections::HashMap<String, String>,
428) -> Result<PipelineSpec, serde_json::Error> {
429 if vars.is_empty() {
430 return Ok(spec.clone());
431 }
432 let mut v = serde_json::to_value(spec)?;
433 walk_subst(&mut v, vars);
434 serde_json::from_value(v)
435}
436
437fn walk_subst(v: &mut serde_json::Value, vars: &std::collections::HashMap<String, String>) {
438 match v {
439 serde_json::Value::String(s) => {
440 *s = subst_str(s, vars);
441 }
442 serde_json::Value::Array(arr) => {
443 for item in arr {
444 walk_subst(item, vars);
445 }
446 }
447 serde_json::Value::Object(obj) => {
448 for (_, val) in obj.iter_mut() {
449 walk_subst(val, vars);
450 }
451 }
452 _ => {}
453 }
454}
455
456fn subst_str(s: &str, vars: &std::collections::HashMap<String, String>) -> String {
457 let mut out = String::with_capacity(s.len());
458 let bytes = s.as_bytes();
459 let mut i = 0;
460 while i < bytes.len() {
461 if i + 1 < bytes.len() && bytes[i] == b'$' && bytes[i + 1] == b'{' {
462 // Buscar el cierre `}`.
463 if let Some(close) = bytes[i + 2..].iter().position(|&b| b == b'}') {
464 let key = std::str::from_utf8(&bytes[i + 2..i + 2 + close]).unwrap_or("");
465 if let Some(val) = vars.get(key) {
466 out.push_str(val);
467 i += 2 + close + 1;
468 continue;
469 }
470 }
471 }
472 out.push(bytes[i] as char);
473 i += 1;
474 }
475 out
476}
477
478#[cfg(test)]
479mod subst_tests {
480 use super::*;
481 use std::collections::HashMap;
482
483 #[test]
484 fn substitute_in_argv_and_label() {
485 let mut vars = HashMap::new();
486 vars.insert("MSG".into(), "hola-mundo".into());
487 vars.insert("LABEL".into(), "renamed".into());
488 let spec = PipelineSpec {
489 label: "p-${LABEL}".into(),
490 workspace: WorkspaceId::new(),
491 nodes: vec![CommandRef {
492 label: "node-${LABEL}".into(),
493 payload: Payload::Native {
494 exec: "/bin/echo".into(),
495 argv: vec!["${MSG}".into()],
496 envp: vec![],
497 },
498 soma: Default::default(),
499 flows: Default::default(),
500 supervision: Supervision::OneShot,
501 }],
502 edges: vec![],
503 discern: DiscernPolicy::default(),
504 restart_on_failure: false,
505 };
506 let out = substitute_vars(&spec, &vars).unwrap();
507 assert_eq!(out.label, "p-renamed");
508 assert_eq!(out.nodes[0].label, "node-renamed");
509 match &out.nodes[0].payload {
510 Payload::Native { argv, .. } => assert_eq!(argv[0], "hola-mundo"),
511 _ => panic!("wrong payload"),
512 }
513 }
514
515 #[test]
516 fn unknown_var_left_intact() {
517 let vars = HashMap::new();
518 let spec = PipelineSpec {
519 label: "p-${UNDEFINED}".into(),
520 workspace: WorkspaceId::new(),
521 nodes: vec![],
522 edges: vec![],
523 discern: DiscernPolicy::default(),
524 restart_on_failure: false,
525 };
526 let out = substitute_vars(&spec, &vars).unwrap();
527 assert_eq!(out.label, "p-${UNDEFINED}");
528 }
529}
530
531#[cfg(test)]
532mod tests {
533 use super::*;
534
535 fn sample_workspace() -> WorkspaceSpec {
536 WorkspaceSpec {
537 label: "demo".into(),
538 soma: SomaSpec::default(),
539 permissions: Permissions::default(),
540 ttl: Some(Duration::from_secs(60)),
541 flow_dirs: vec![FlowSlot {
542 name: "out".into(),
543 direction: FlowDirection::Output,
544 scope: FlowScope::Public,
545 }],
546 on_exit: ExitPolicy::Reap,
547 quota_enforce: Default::default(),
548 }
549 }
550
551 #[test]
552 fn workspace_toml_roundtrip() {
553 let ws = sample_workspace();
554 let s = toml::to_string(&ws).unwrap();
555 let back: WorkspaceSpec = toml::from_str(&s).unwrap();
556 assert_eq!(back.label, ws.label);
557 assert_eq!(back.ttl, ws.ttl);
558 assert_eq!(back.flow_dirs.len(), 1);
559 }
560
561 #[test]
562 fn workspace_json_roundtrip() {
563 let ws = sample_workspace();
564 let s = serde_json::to_string(&ws).unwrap();
565 let back: WorkspaceSpec = serde_json::from_str(&s).unwrap();
566 assert_eq!(back.label, ws.label);
567 }
568
569 #[test]
570 fn workspace_compiles_to_card() {
571 let ws = sample_workspace();
572 let id = WorkspaceId::new();
573 let c = ws.to_card(id).unwrap();
574 assert_eq!(c.id, id.0);
575 assert!(c.label.starts_with("shipote.workspace."));
576 assert!(matches!(c.payload, Payload::Virtual));
577 }
578
579 #[test]
580 fn empty_label_rejected() {
581 let mut ws = sample_workspace();
582 ws.label = String::new();
583 assert!(ws.to_card(WorkspaceId::new()).is_err());
584 }
585
586 #[test]
587 fn pipeline_validates_edges() {
588 let p = PipelineSpec {
589 label: "p".into(),
590 workspace: WorkspaceId::new(),
591 nodes: vec![CommandRef {
592 label: "a".into(),
593 payload: Payload::Virtual,
594 soma: SomaSpec::default(),
595 flows: brahman_card::Flows::default(),
596 supervision: Supervision::OneShot,
597 }],
598 edges: vec![FlowEdge {
599 from: 0,
600 from_output: "x".into(),
601 to: 5,
602 to_input: "y".into(),
603 }],
604 discern: DiscernPolicy::default(),
605 restart_on_failure: false,
606 };
607 assert!(p.validate().is_err());
608 }
609
610 #[test]
611 fn intersect_soma_takes_more_restrictive() {
612 let mut child = SomaSpec::default();
613 child.rlimits.mem_bytes = Some(1_000_000);
614 let mut ws = SomaSpec::default();
615 ws.rlimits.mem_bytes = Some(500_000);
616 ws.namespaces.user = true;
617 let r = intersect_soma(&child, &ws);
618 assert_eq!(r.rlimits.mem_bytes, Some(500_000));
619 assert!(r.namespaces.user);
620 }
621}