⎇ c22d2480b93b7802be8e8ea5a0075b8175c4fedc
crates/modules/shipote/shipote-card/src/lib.rs
History
Who touched what
raw
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
99#[derive(Debug, Clone, Serialize, Deserialize)]
100pub struct FlowSlot {
101 pub name: String,
102 pub direction: FlowDirection,
103 /// Si `Workspace`, sólo otros nodos del mismo workspace pueden empatar.
104 /// Si `Public`, el broker global puede emparejar.
105 #[serde(default)]
106 pub scope: FlowScope,
107}
108
109#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
110#[serde(rename_all = "lowercase")]
111pub enum FlowDirection {
112 Input,
113 Output,
114}
115
116#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
117#[serde(rename_all = "lowercase")]
118pub enum FlowScope {
119 #[default]
120 Workspace,
121 Public,
122}
123
124#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
125#[serde(rename_all = "kebab-case")]
126pub enum ExitPolicy {
127 /// Reapear procesos hijos y descartar estado.
128 #[default]
129 Reap,
130 /// Mantener el workspace en `stopped` para inspección.
131 Keep,
132 /// Tomar snapshot del estado (para restart posterior).
133 Snapshot,
134}
135
136mod opt_duration_millis {
137 use serde::{Deserialize, Deserializer, Serialize, Serializer};
138 use std::time::Duration;
139
140 pub fn serialize<S: Serializer>(d: &Option<Duration>, s: S) -> Result<S::Ok, S::Error> {
141 d.map(|x| x.as_millis() as u64).serialize(s)
142 }
143
144 pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Option<Duration>, D::Error> {
145 let v: Option<u64> = Option::deserialize(d)?;
146 Ok(v.map(Duration::from_millis))
147 }
148}
149
150// =====================================================================
151// CommandRef
152// =====================================================================
153
154/// Un comando que vive dentro de un workspace. Se compila a una `Card` con
155/// `pin_to` apuntando al workspace padre (label) y su `SomaSpec`
156/// intersectado con el del workspace.
157#[derive(Debug, Clone, Serialize, Deserialize)]
158pub struct CommandRef {
159 pub label: String,
160 pub payload: Payload,
161
162 /// SomaSpec del comando. El compilador lo intersecta con el del workspace.
163 #[serde(default)]
164 pub soma: SomaSpec,
165
166 /// Inputs/outputs tipados (mismos `Flow` de brahman-card).
167 #[serde(default)]
168 pub flows: brahman_card::Flows,
169
170 /// Política de supervisión. Default `OneShot` (un comando se ejecuta y muere).
171 #[serde(default = "default_oneshot")]
172 pub supervision: Supervision,
173}
174
175fn default_oneshot() -> Supervision {
176 Supervision::OneShot
177}
178
179// =====================================================================
180// Pipeline
181// =====================================================================
182
183#[derive(Debug, Clone, Serialize, Deserialize)]
184pub struct PipelineSpec {
185 pub label: String,
186 pub workspace: WorkspaceId,
187 pub nodes: Vec<CommandRef>,
188 #[serde(default)]
189 pub edges: Vec<FlowEdge>,
190 #[serde(default)]
191 pub discern: DiscernPolicy,
192}
193
194#[derive(Debug, Clone, Serialize, Deserialize)]
195pub struct FlowEdge {
196 /// Índice en `PipelineSpec.nodes` del productor.
197 pub from: usize,
198 /// Nombre del Flow output del productor.
199 pub from_output: String,
200 /// Índice en `PipelineSpec.nodes` del consumidor.
201 pub to: usize,
202 /// Nombre del Flow input del consumidor.
203 pub to_input: String,
204}
205
206#[derive(Debug, Clone, Default, Serialize, Deserialize)]
207pub struct DiscernPolicy {
208 /// Bytes a samplear por flow para el discernidor. Default 4 KiB.
209 #[serde(default = "default_sample_bytes")]
210 pub sample_bytes: usize,
211 /// Si `true`, enriquece la Card del producer con el TypeRef detectado.
212 #[serde(default = "default_true")]
213 pub enrich_producer: bool,
214}
215
216fn default_sample_bytes() -> usize {
217 4096
218}
219fn default_true() -> bool {
220 true
221}
222
223// =====================================================================
224// Compilación a Card
225// =====================================================================
226
227#[derive(Debug, Error)]
228pub enum CompileError {
229 #[error("workspace label vacío")]
230 EmptyWorkspaceLabel,
231 #[error("comando con label vacío en posición {0}")]
232 EmptyCommandLabel(usize),
233 #[error("edge fuera de rango: from={from}, to={to}, nodes={nodes}")]
234 EdgeOutOfBounds { from: usize, to: usize, nodes: usize },
235}
236
237impl WorkspaceSpec {
238 /// Compila el WorkspaceSpec a una Card raíz que el Incarnator puede
239 /// encarnar. Usa `Payload::Virtual` (el workspace no es un proceso por
240 /// sí solo; sólo aloja hijos).
241 pub fn to_card(&self, id: WorkspaceId) -> Result<Card, CompileError> {
242 if self.label.trim().is_empty() {
243 return Err(CompileError::EmptyWorkspaceLabel);
244 }
245 let mut c = Card::new(format!("shipote.workspace.{}", self.label));
246 c.id = id.0;
247 c.soma = self.soma.clone();
248 c.permissions = self.permissions.clone();
249 c.payload = Payload::Virtual;
250 c.supervision = Supervision::OneShot;
251 Ok(c)
252 }
253}
254
255impl CommandRef {
256 /// Compila un CommandRef a Card hija de un workspace. La Card resultante
257 /// referencia al workspace por label en `pin_to` de cada Flow.
258 pub fn to_card(&self, idx: usize, workspace_label: &str) -> Result<Card, CompileError> {
259 if self.label.trim().is_empty() {
260 return Err(CompileError::EmptyCommandLabel(idx));
261 }
262 let mut c = Card::new(format!("shipote.cmd.{}.{}", workspace_label, self.label));
263 c.payload = self.payload.clone();
264 c.soma = intersect_soma(&self.soma, /*workspace*/ &SomaSpec::default());
265 c.supervision = self.supervision.clone();
266 c.flow = self.flows.clone();
267 // pin_to del workspace en cada Flow input/output → el broker prefiere
268 // resolver dentro del mismo workspace cuando hay candidatos múltiples.
269 let pin = format!("shipote.workspace.{}", workspace_label);
270 for f in c.flow.input.iter_mut().chain(c.flow.output.iter_mut()) {
271 if f.pin_to.is_none() {
272 f.pin_to = Some(pin.clone());
273 }
274 }
275 Ok(c)
276 }
277}
278
279/// Intersección conservadora: si el workspace pidió aislamiento, la hija
280/// también lo tiene (no puede aflojar). Si la hija pidió aislamiento extra,
281/// se respeta.
282fn intersect_soma(child: &SomaSpec, ws: &SomaSpec) -> SomaSpec {
283 let mut out = child.clone();
284 out.namespaces.mount |= ws.namespaces.mount;
285 out.namespaces.pid |= ws.namespaces.pid;
286 out.namespaces.net |= ws.namespaces.net;
287 out.namespaces.uts |= ws.namespaces.uts;
288 out.namespaces.ipc |= ws.namespaces.ipc;
289 out.namespaces.user |= ws.namespaces.user;
290 out.namespaces.cgroup |= ws.namespaces.cgroup;
291 // rlimits: el menor (más restrictivo) gana.
292 out.rlimits.mem_bytes = min_opt(out.rlimits.mem_bytes, ws.rlimits.mem_bytes);
293 out.rlimits.nproc = min_opt(out.rlimits.nproc, ws.rlimits.nproc);
294 out.rlimits.nofile = min_opt(out.rlimits.nofile, ws.rlimits.nofile);
295 out
296}
297
298fn min_opt<T: Ord + Copy>(a: Option<T>, b: Option<T>) -> Option<T> {
299 match (a, b) {
300 (Some(x), Some(y)) => Some(x.min(y)),
301 (Some(x), None) | (None, Some(x)) => Some(x),
302 (None, None) => None,
303 }
304}
305
306impl PipelineSpec {
307 pub fn validate(&self) -> Result<(), CompileError> {
308 let n = self.nodes.len();
309 for (i, c) in self.nodes.iter().enumerate() {
310 if c.label.trim().is_empty() {
311 return Err(CompileError::EmptyCommandLabel(i));
312 }
313 }
314 for e in &self.edges {
315 if e.from >= n || e.to >= n {
316 return Err(CompileError::EdgeOutOfBounds {
317 from: e.from,
318 to: e.to,
319 nodes: n,
320 });
321 }
322 }
323 Ok(())
324 }
325}
326
327// =====================================================================
328// I/O conveniencia (TOML + JSON)
329// =====================================================================
330
331#[derive(Debug, Error)]
332pub enum LoadError {
333 #[error("io: {0}")]
334 Io(#[from] std::io::Error),
335 #[error("toml: {0}")]
336 Toml(#[from] toml::de::Error),
337 #[error("json: {0}")]
338 Json(#[from] serde_json::Error),
339 #[error("formato desconocido (esperado .toml o .json)")]
340 UnknownFormat,
341}
342
343pub fn load_workspace_spec(path: &std::path::Path) -> Result<WorkspaceSpec, LoadError> {
344 let raw = std::fs::read_to_string(path)?;
345 match path.extension().and_then(|s| s.to_str()) {
346 Some("toml") => Ok(toml::from_str(&raw)?),
347 Some("json") => Ok(serde_json::from_str(&raw)?),
348 _ => Err(LoadError::UnknownFormat),
349 }
350}
351
352pub fn load_pipeline_spec(path: &std::path::Path) -> Result<PipelineSpec, LoadError> {
353 let raw = std::fs::read_to_string(path)?;
354 match path.extension().and_then(|s| s.to_str()) {
355 Some("toml") => Ok(toml::from_str(&raw)?),
356 Some("json") => Ok(serde_json::from_str(&raw)?),
357 _ => Err(LoadError::UnknownFormat),
358 }
359}
360
361#[cfg(test)]
362mod tests {
363 use super::*;
364
365 fn sample_workspace() -> WorkspaceSpec {
366 WorkspaceSpec {
367 label: "demo".into(),
368 soma: SomaSpec::default(),
369 permissions: Permissions::default(),
370 ttl: Some(Duration::from_secs(60)),
371 flow_dirs: vec![FlowSlot {
372 name: "out".into(),
373 direction: FlowDirection::Output,
374 scope: FlowScope::Public,
375 }],
376 on_exit: ExitPolicy::Reap,
377 }
378 }
379
380 #[test]
381 fn workspace_toml_roundtrip() {
382 let ws = sample_workspace();
383 let s = toml::to_string(&ws).unwrap();
384 let back: WorkspaceSpec = toml::from_str(&s).unwrap();
385 assert_eq!(back.label, ws.label);
386 assert_eq!(back.ttl, ws.ttl);
387 assert_eq!(back.flow_dirs.len(), 1);
388 }
389
390 #[test]
391 fn workspace_json_roundtrip() {
392 let ws = sample_workspace();
393 let s = serde_json::to_string(&ws).unwrap();
394 let back: WorkspaceSpec = serde_json::from_str(&s).unwrap();
395 assert_eq!(back.label, ws.label);
396 }
397
398 #[test]
399 fn workspace_compiles_to_card() {
400 let ws = sample_workspace();
401 let id = WorkspaceId::new();
402 let c = ws.to_card(id).unwrap();
403 assert_eq!(c.id, id.0);
404 assert!(c.label.starts_with("shipote.workspace."));
405 assert!(matches!(c.payload, Payload::Virtual));
406 }
407
408 #[test]
409 fn empty_label_rejected() {
410 let mut ws = sample_workspace();
411 ws.label = String::new();
412 assert!(ws.to_card(WorkspaceId::new()).is_err());
413 }
414
415 #[test]
416 fn pipeline_validates_edges() {
417 let p = PipelineSpec {
418 label: "p".into(),
419 workspace: WorkspaceId::new(),
420 nodes: vec![CommandRef {
421 label: "a".into(),
422 payload: Payload::Virtual,
423 soma: SomaSpec::default(),
424 flows: brahman_card::Flows::default(),
425 supervision: Supervision::OneShot,
426 }],
427 edges: vec![FlowEdge {
428 from: 0,
429 from_output: "x".into(),
430 to: 5,
431 to_input: "y".into(),
432 }],
433 discern: DiscernPolicy::default(),
434 };
435 assert!(p.validate().is_err());
436 }
437
438 #[test]
439 fn intersect_soma_takes_more_restrictive() {
440 let mut child = SomaSpec::default();
441 child.rlimits.mem_bytes = Some(1_000_000);
442 let mut ws = SomaSpec::default();
443 ws.rlimits.mem_bytes = Some(500_000);
444 ws.namespaces.user = true;
445 let r = intersect_soma(&child, &ws);
446 assert_eq!(r.rlimits.mem_bytes, Some(500_000));
447 assert!(r.namespaces.user);
448 }
449}