⎇ 72758e75ce515f355f27615671c95ecaf8c2a8bc
crates/modules/shipote/shipote-card/src/lib.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
feat(shipote): throughput card + rate-limit + snapshot incremental (fase Q)
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 /// Backoff inicial entre restarts (ms). Crece exponencialmente
226 /// hasta `restart_max_backoff_ms`. Default 200ms = ~5 restarts/s
227 /// inicial, escalando rápido.
228 #[serde(default = "default_restart_backoff")]
229 pub restart_backoff_ms: u64,
230 /// Backoff máximo (ms). Default 30s. El backoff no crece más allá.
231 #[serde(default = "default_restart_max_backoff")]
232 pub restart_max_backoff_ms: u64,
233 /// Máximo de restarts antes de dar up. `0` = infinito. Default 0.
234 /// Útil para fail-loud cuando un pipeline siempre falla.
235 #[serde(default)]
236 pub restart_max: u32,
237}
238
239fn default_restart_backoff() -> u64 {
240 200
241}
242fn default_restart_max_backoff() -> u64 {
243 30_000
244}
245
246#[derive(Debug, Clone, Serialize, Deserialize)]
247pub struct FlowEdge {
248 /// Índice en `PipelineSpec.nodes` del productor.
249 pub from: usize,
250 /// Nombre del Flow output del productor.
251 pub from_output: String,
252 /// Índice en `PipelineSpec.nodes` del consumidor.
253 pub to: usize,
254 /// Nombre del Flow input del consumidor.
255 pub to_input: String,
256}
257
258#[derive(Debug, Clone, Serialize, Deserialize)]
259pub struct DiscernPolicy {
260 /// Bytes a samplear por flow para el discernidor. Default 4 KiB.
261 #[serde(default = "default_sample_bytes")]
262 pub sample_bytes: usize,
263 /// Si `true`, enriquece la Card del producer con el TypeRef detectado.
264 #[serde(default = "default_true")]
265 pub enrich_producer: bool,
266 /// Chunks que el FlowChannel guarda en replay buffer para subscribers
267 /// tarde. Default 32. Subir si los productores escriben en ráfagas y
268 /// querés que los consumidores tardíos vean toda la salida.
269 #[serde(default = "default_replay_chunks")]
270 pub replay_chunks: usize,
271 /// Tope adicional por **bytes** acumulados en el replay buffer. Lo
272 /// que se exceda primero (chunks o bytes) drop-ea el chunk más viejo.
273 /// `0` = sin tope por bytes (sólo aplica `replay_chunks`). Útil para
274 /// productores con chunks de tamaño variable.
275 #[serde(default)]
276 pub replay_bytes: usize,
277 /// Rate-limit del flow channel (bytes/s). `0` = sin límite. Si está
278 /// definido, el splitter sleeps proporcional al tamaño del chunk
279 /// antes de re-broadcastear. Protege subscribers lentos.
280 #[serde(default)]
281 pub max_bytes_per_sec: u64,
282}
283
284impl Default for DiscernPolicy {
285 fn default() -> Self {
286 Self {
287 sample_bytes: default_sample_bytes(),
288 enrich_producer: default_true(),
289 replay_chunks: default_replay_chunks(),
290 replay_bytes: 0,
291 max_bytes_per_sec: 0,
292 }
293 }
294}
295
296fn default_sample_bytes() -> usize {
297 4096
298}
299fn default_true() -> bool {
300 true
301}
302fn default_replay_chunks() -> usize {
303 32
304}
305
306// =====================================================================
307// Compilación a Card
308// =====================================================================
309
310#[derive(Debug, Error)]
311pub enum CompileError {
312 #[error("workspace label vacío")]
313 EmptyWorkspaceLabel,
314 #[error("comando con label vacío en posición {0}")]
315 EmptyCommandLabel(usize),
316 #[error("edge fuera de rango: from={from}, to={to}, nodes={nodes}")]
317 EdgeOutOfBounds { from: usize, to: usize, nodes: usize },
318}
319
320impl WorkspaceSpec {
321 /// Compila el WorkspaceSpec a una Card raíz que el Incarnator puede
322 /// encarnar. Usa `Payload::Virtual` (el workspace no es un proceso por
323 /// sí solo; sólo aloja hijos).
324 pub fn to_card(&self, id: WorkspaceId) -> Result<Card, CompileError> {
325 if self.label.trim().is_empty() {
326 return Err(CompileError::EmptyWorkspaceLabel);
327 }
328 let mut c = Card::new(format!("shipote.workspace.{}", self.label));
329 c.id = id.0;
330 c.soma = self.soma.clone();
331 c.permissions = self.permissions.clone();
332 c.payload = Payload::Virtual;
333 c.supervision = Supervision::OneShot;
334 Ok(c)
335 }
336}
337
338impl CommandRef {
339 /// Compila un CommandRef a Card hija de un workspace. La Card resultante
340 /// referencia al workspace por label en `pin_to` de cada Flow.
341 pub fn to_card(&self, idx: usize, workspace_label: &str) -> Result<Card, CompileError> {
342 if self.label.trim().is_empty() {
343 return Err(CompileError::EmptyCommandLabel(idx));
344 }
345 let mut c = Card::new(format!("shipote.cmd.{}.{}", workspace_label, self.label));
346 c.payload = self.payload.clone();
347 c.soma = intersect_soma(&self.soma, /*workspace*/ &SomaSpec::default());
348 c.supervision = self.supervision.clone();
349 c.flow = self.flows.clone();
350 // pin_to del workspace en cada Flow input/output → el broker prefiere
351 // resolver dentro del mismo workspace cuando hay candidatos múltiples.
352 let pin = format!("shipote.workspace.{}", workspace_label);
353 for f in c.flow.input.iter_mut().chain(c.flow.output.iter_mut()) {
354 if f.pin_to.is_none() {
355 f.pin_to = Some(pin.clone());
356 }
357 }
358 Ok(c)
359 }
360}
361
362/// Intersección conservadora: si el workspace pidió aislamiento, la hija
363/// también lo tiene (no puede aflojar). Si la hija pidió aislamiento extra,
364/// se respeta.
365fn intersect_soma(child: &SomaSpec, ws: &SomaSpec) -> SomaSpec {
366 let mut out = child.clone();
367 out.namespaces.mount |= ws.namespaces.mount;
368 out.namespaces.pid |= ws.namespaces.pid;
369 out.namespaces.net |= ws.namespaces.net;
370 out.namespaces.uts |= ws.namespaces.uts;
371 out.namespaces.ipc |= ws.namespaces.ipc;
372 out.namespaces.user |= ws.namespaces.user;
373 out.namespaces.cgroup |= ws.namespaces.cgroup;
374 // rlimits: el menor (más restrictivo) gana.
375 out.rlimits.mem_bytes = min_opt(out.rlimits.mem_bytes, ws.rlimits.mem_bytes);
376 out.rlimits.nproc = min_opt(out.rlimits.nproc, ws.rlimits.nproc);
377 out.rlimits.nofile = min_opt(out.rlimits.nofile, ws.rlimits.nofile);
378 out
379}
380
381fn min_opt<T: Ord + Copy>(a: Option<T>, b: Option<T>) -> Option<T> {
382 match (a, b) {
383 (Some(x), Some(y)) => Some(x.min(y)),
384 (Some(x), None) | (None, Some(x)) => Some(x),
385 (None, None) => None,
386 }
387}
388
389impl PipelineSpec {
390 pub fn validate(&self) -> Result<(), CompileError> {
391 let n = self.nodes.len();
392 for (i, c) in self.nodes.iter().enumerate() {
393 if c.label.trim().is_empty() {
394 return Err(CompileError::EmptyCommandLabel(i));
395 }
396 }
397 for e in &self.edges {
398 if e.from >= n || e.to >= n {
399 return Err(CompileError::EdgeOutOfBounds {
400 from: e.from,
401 to: e.to,
402 nodes: n,
403 });
404 }
405 }
406 Ok(())
407 }
408}
409
410// =====================================================================
411// I/O conveniencia (TOML + JSON)
412// =====================================================================
413
414#[derive(Debug, Error)]
415pub enum LoadError {
416 #[error("io: {0}")]
417 Io(#[from] std::io::Error),
418 #[error("toml: {0}")]
419 Toml(#[from] toml::de::Error),
420 #[error("json: {0}")]
421 Json(#[from] serde_json::Error),
422 #[error("formato desconocido (esperado .toml o .json)")]
423 UnknownFormat,
424}
425
426pub fn load_workspace_spec(path: &std::path::Path) -> Result<WorkspaceSpec, LoadError> {
427 let raw = std::fs::read_to_string(path)?;
428 match path.extension().and_then(|s| s.to_str()) {
429 Some("toml") => Ok(toml::from_str(&raw)?),
430 Some("json") => Ok(serde_json::from_str(&raw)?),
431 _ => Err(LoadError::UnknownFormat),
432 }
433}
434
435pub fn load_pipeline_spec(path: &std::path::Path) -> Result<PipelineSpec, LoadError> {
436 let raw = std::fs::read_to_string(path)?;
437 match path.extension().and_then(|s| s.to_str()) {
438 Some("toml") => Ok(toml::from_str(&raw)?),
439 Some("json") => Ok(serde_json::from_str(&raw)?),
440 _ => Err(LoadError::UnknownFormat),
441 }
442}
443
444/// Sustituye `${KEY}` en todos los strings del spec por el valor de
445/// `vars["KEY"]`. Variables sin match quedan intactas (no se borra el
446/// placeholder — útil para detectar olvidos).
447///
448/// Walk recursivo sobre la representación JSON intermedia para cubrir
449/// labels, argv, envp, paths y cualquier String del schema.
450pub fn substitute_vars(
451 spec: &PipelineSpec,
452 vars: &std::collections::HashMap<String, String>,
453) -> Result<PipelineSpec, serde_json::Error> {
454 if vars.is_empty() {
455 return Ok(spec.clone());
456 }
457 let mut v = serde_json::to_value(spec)?;
458 walk_subst(&mut v, vars);
459 serde_json::from_value(v)
460}
461
462fn walk_subst(v: &mut serde_json::Value, vars: &std::collections::HashMap<String, String>) {
463 match v {
464 serde_json::Value::String(s) => {
465 *s = subst_str(s, vars);
466 }
467 serde_json::Value::Array(arr) => {
468 for item in arr {
469 walk_subst(item, vars);
470 }
471 }
472 serde_json::Value::Object(obj) => {
473 for (_, val) in obj.iter_mut() {
474 walk_subst(val, vars);
475 }
476 }
477 _ => {}
478 }
479}
480
481fn subst_str(s: &str, vars: &std::collections::HashMap<String, String>) -> String {
482 let mut out = String::with_capacity(s.len());
483 let bytes = s.as_bytes();
484 let mut i = 0;
485 while i < bytes.len() {
486 if i + 1 < bytes.len() && bytes[i] == b'$' && bytes[i + 1] == b'{' {
487 // Buscar el cierre `}`.
488 if let Some(close) = bytes[i + 2..].iter().position(|&b| b == b'}') {
489 let key = std::str::from_utf8(&bytes[i + 2..i + 2 + close]).unwrap_or("");
490 if let Some(val) = vars.get(key) {
491 out.push_str(val);
492 i += 2 + close + 1;
493 continue;
494 }
495 }
496 }
497 out.push(bytes[i] as char);
498 i += 1;
499 }
500 out
501}
502
503#[cfg(test)]
504mod subst_tests {
505 use super::*;
506 use std::collections::HashMap;
507
508 #[test]
509 fn substitute_in_argv_and_label() {
510 let mut vars = HashMap::new();
511 vars.insert("MSG".into(), "hola-mundo".into());
512 vars.insert("LABEL".into(), "renamed".into());
513 let spec = PipelineSpec {
514 label: "p-${LABEL}".into(),
515 workspace: WorkspaceId::new(),
516 nodes: vec![CommandRef {
517 label: "node-${LABEL}".into(),
518 payload: Payload::Native {
519 exec: "/bin/echo".into(),
520 argv: vec!["${MSG}".into()],
521 envp: vec![],
522 },
523 soma: Default::default(),
524 flows: Default::default(),
525 supervision: Supervision::OneShot,
526 }],
527 edges: vec![],
528 discern: DiscernPolicy::default(),
529 restart_on_failure: false,
530 restart_backoff_ms: 200,
531 restart_max_backoff_ms: 30_000,
532 restart_max: 0,
533 };
534 let out = substitute_vars(&spec, &vars).unwrap();
535 assert_eq!(out.label, "p-renamed");
536 assert_eq!(out.nodes[0].label, "node-renamed");
537 match &out.nodes[0].payload {
538 Payload::Native { argv, .. } => assert_eq!(argv[0], "hola-mundo"),
539 _ => panic!("wrong payload"),
540 }
541 }
542
543 #[test]
544 fn unknown_var_left_intact() {
545 let vars = HashMap::new();
546 let spec = PipelineSpec {
547 label: "p-${UNDEFINED}".into(),
548 workspace: WorkspaceId::new(),
549 nodes: vec![],
550 edges: vec![],
551 discern: DiscernPolicy::default(),
552 restart_on_failure: false,
553 restart_backoff_ms: 200,
554 restart_max_backoff_ms: 30_000,
555 restart_max: 0,
556 };
557 let out = substitute_vars(&spec, &vars).unwrap();
558 assert_eq!(out.label, "p-${UNDEFINED}");
559 }
560}
561
562#[cfg(test)]
563mod tests {
564 use super::*;
565
566 fn sample_workspace() -> WorkspaceSpec {
567 WorkspaceSpec {
568 label: "demo".into(),
569 soma: SomaSpec::default(),
570 permissions: Permissions::default(),
571 ttl: Some(Duration::from_secs(60)),
572 flow_dirs: vec![FlowSlot {
573 name: "out".into(),
574 direction: FlowDirection::Output,
575 scope: FlowScope::Public,
576 }],
577 on_exit: ExitPolicy::Reap,
578 quota_enforce: Default::default(),
579 }
580 }
581
582 #[test]
583 fn workspace_toml_roundtrip() {
584 let ws = sample_workspace();
585 let s = toml::to_string(&ws).unwrap();
586 let back: WorkspaceSpec = toml::from_str(&s).unwrap();
587 assert_eq!(back.label, ws.label);
588 assert_eq!(back.ttl, ws.ttl);
589 assert_eq!(back.flow_dirs.len(), 1);
590 }
591
592 #[test]
593 fn workspace_json_roundtrip() {
594 let ws = sample_workspace();
595 let s = serde_json::to_string(&ws).unwrap();
596 let back: WorkspaceSpec = serde_json::from_str(&s).unwrap();
597 assert_eq!(back.label, ws.label);
598 }
599
600 #[test]
601 fn workspace_compiles_to_card() {
602 let ws = sample_workspace();
603 let id = WorkspaceId::new();
604 let c = ws.to_card(id).unwrap();
605 assert_eq!(c.id, id.0);
606 assert!(c.label.starts_with("shipote.workspace."));
607 assert!(matches!(c.payload, Payload::Virtual));
608 }
609
610 #[test]
611 fn empty_label_rejected() {
612 let mut ws = sample_workspace();
613 ws.label = String::new();
614 assert!(ws.to_card(WorkspaceId::new()).is_err());
615 }
616
617 #[test]
618 fn pipeline_validates_edges() {
619 let p = PipelineSpec {
620 label: "p".into(),
621 workspace: WorkspaceId::new(),
622 nodes: vec![CommandRef {
623 label: "a".into(),
624 payload: Payload::Virtual,
625 soma: SomaSpec::default(),
626 flows: brahman_card::Flows::default(),
627 supervision: Supervision::OneShot,
628 }],
629 edges: vec![FlowEdge {
630 from: 0,
631 from_output: "x".into(),
632 to: 5,
633 to_input: "y".into(),
634 }],
635 discern: DiscernPolicy::default(),
636 restart_on_failure: false,
637 restart_backoff_ms: 200,
638 restart_max_backoff_ms: 30_000,
639 restart_max: 0,
640 };
641 assert!(p.validate().is_err());
642 }
643
644 #[test]
645 fn intersect_soma_takes_more_restrictive() {
646 let mut child = SomaSpec::default();
647 child.rlimits.mem_bytes = Some(1_000_000);
648 let mut ws = SomaSpec::default();
649 ws.rlimits.mem_bytes = Some(500_000);
650 ws.namespaces.user = true;
651 let r = intersect_soma(&child, &ws);
652 assert_eq!(r.rlimits.mem_bytes, Some(500_000));
653 assert!(r.namespaces.user);
654 }
655}