⎇ c3f9c9e36a758ad07434939dfa9bb214aff4acfb
crates/apps/shipote-daemon/src/main.rs
Historia
Quién tocó qué
crudo
feat(shipote): pipeline backoff + quota card + logs follow (fase M)
1//! `shipote-daemon` — punto de entrada del runtime de shipote.
2//!
3//! Responsabilidades:
4//! - Escuchar el Unix socket admin (default: `$XDG_RUNTIME_DIR/shipote.sock`).
5//! - Despachar mensajes del [`shipote_protocol`] al [`WorkspaceManager`].
6//! - Reapear hijos periódicamente.
7//!
8//! Lo que NO hace en v1:
9//! - Sidecar al broker / handshake con Init (futuro: cuando un workspace
10//! exponga `service_socket`, anunciar al broker).
11//! - GUI (futuro `shipote-shell` con yahweh_launcher).
12
13use anyhow::Context;
14use brahman_card::{Card, CardKind, Flow, Flows, Lifecycle, Payload, Supervision, TypeRef};
15use ente_incarnate::IncarnatorConfig;
16use shipote_core::WorkspaceManager;
17use shipote_discern::{DiscernPipeline, Hint};
18use shipote_protocol::{
19 default_socket_path, read_frame, write_frame, CommandInfo as ProtoCommandInfo,
20 EdgeDiscernmentInfo, FlowInfo, QuotaReportInfo, Request, Response, WorkspaceStatsInfo,
21 WorkspaceSummary,
22};
23use std::sync::Arc;
24use tokio::net::{UnixListener, UnixStream};
25use tracing::{error, info, warn};
26
27#[tokio::main]
28async fn main() -> anyhow::Result<()> {
29 init_tracing();
30 let sock = default_socket_path();
31 if sock.exists() {
32 // Si ya existe, asumimos restart limpio. Si hubiera otro daemon vivo,
33 // bind fallaría con EADDRINUSE — más adelante: lockfile + check de PID.
34 let _ = std::fs::remove_file(&sock);
35 }
36 if let Some(parent) = sock.parent() {
37 let _ = std::fs::create_dir_all(parent);
38 }
39 let listener = UnixListener::bind(&sock).with_context(|| format!("bind {}", sock.display()))?;
40 info!(socket = %sock.display(), "shipote-daemon listening");
41
42 // Sidecar pool: una sesión global del daemon + N sesiones efímeras
43 // por edge enriquecido tras cada pipeline tap.
44 let sidecar_pool = match brahman_sidecar::SidecarPool::new() {
45 Ok(p) => Some(Arc::new(p)),
46 Err(e) => {
47 warn!(?e, "SidecarPool falló — broker integration disabled");
48 None
49 }
50 };
51 if let Some(pool) = &sidecar_pool {
52 pool.spawn(build_daemon_card(&sock));
53 }
54
55 let mgr = Arc::new(WorkspaceManager::new(IncarnatorConfig {
56 // El daemon aún no se conecta al broker; cuando lo haga, este path
57 // se llenará desde el handshake.
58 bus_sock: None,
59 notify_socket: None,
60 extra_env: vec![("SHIPOTE_DAEMON".into(), "1".into())],
61 // strict_caps=false en v1: queremos UX permisiva (correr en non-root
62 // sin user_ns y avisar via warnings, no abortar).
63 strict_caps: false,
64 }));
65
66 // Restaurar snapshot previo si existe. Workspaces se recrean; los
67 // pids de comandos viejos NO se recuperan (kernel los mató).
68 let snapshot_path = shipote_core::persist::default_snapshot_path();
69 if let Err(e) = mgr.restore_snapshot(&snapshot_path).await {
70 warn!(?e, "restore_snapshot falló — start fresh");
71 }
72
73 // Save-on-shutdown via SIGTERM/SIGINT handler. tokio::signal soporta
74 // ambos en Linux.
75 {
76 let mgr = mgr.clone();
77 let path = snapshot_path.clone();
78 tokio::spawn(async move {
79 let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
80 .expect("SIGTERM handler");
81 let mut int = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
82 .expect("SIGINT handler");
83 tokio::select! {
84 _ = term.recv() => info!("SIGTERM — saving snapshot"),
85 _ = int.recv() => info!("SIGINT — saving snapshot"),
86 }
87 if let Err(e) = mgr.save_snapshot(&path).await {
88 warn!(?e, "save_snapshot falló");
89 }
90 std::process::exit(0);
91 });
92 }
93
94 let discerner = Arc::new(DiscernPipeline::default_pipeline());
95
96 // Reaper periódico cada 500 ms. Además drena pipelines pendientes
97 // de restart (supervisión a nivel pipeline).
98 {
99 let mgr = mgr.clone();
100 tokio::spawn(async move {
101 let mut tick = tokio::time::interval(std::time::Duration::from_millis(500));
102 loop {
103 tick.tick().await;
104 mgr.reap_dead().await;
105 let pending = mgr.take_pending_restarts().await;
106 for sup in pending {
107 let backoff = std::time::Duration::from_millis(sup.current_backoff_ms);
108 info!(
109 label = %sup.spec.label,
110 restart_count = sup.restart_count,
111 backoff_ms = sup.current_backoff_ms,
112 "pipeline restart: relaunching after backoff"
113 );
114 // Backoff antes del relaunch — anti-thrash.
115 tokio::time::sleep(backoff).await;
116 let inc = mgr.incarnator_handle();
117 let disc = std::sync::Arc::new(DiscernPipeline::default_pipeline());
118 let workspace = sup.spec.workspace;
119 let ws_label = mgr.workspace_label(workspace).await.unwrap_or_default();
120 let tap = sup.tap;
121 let mut new_spec = sup.spec.clone();
122 new_spec.restart_on_failure = true;
123 // Escalar el backoff para la PRÓXIMA falla.
124 let next_backoff = (sup.current_backoff_ms * 2)
125 .min(new_spec.restart_max_backoff_ms);
126 match shipote_core::pipeline::run_pipeline(
127 &new_spec,
128 &ws_label,
129 tap,
130 disc,
131 inc,
132 Some(mgr.clone()),
133 )
134 .await
135 {
136 Ok(launch) => {
137 mgr.register_pipeline_commands(
138 workspace,
139 launch.pipeline,
140 launch.command_pids.clone(),
141 )
142 .await;
143 // Re-registrar supervisor con backoff escalado +
144 // restart_count preservado.
145 mgr.register_pipeline_supervisor_with_state(
146 launch.pipeline,
147 workspace,
148 new_spec,
149 tap,
150 sup.restart_count,
151 next_backoff,
152 )
153 .await;
154 }
155 Err(e) => {
156 warn!(?e, "pipeline restart failed");
157 }
158 }
159 }
160 }
161 });
162 }
163
164 loop {
165 match listener.accept().await {
166 Ok((stream, _)) => {
167 let mgr = mgr.clone();
168 let disc = discerner.clone();
169 let pool = sidecar_pool.clone();
170 tokio::spawn(async move {
171 if let Err(e) = handle_client(stream, mgr, disc, pool).await {
172 warn!(?e, "client handler error");
173 }
174 });
175 }
176 Err(e) => {
177 error!(?e, "accept failed");
178 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
179 }
180 }
181 }
182}
183
184async fn handle_client(
185 mut stream: UnixStream,
186 mgr: Arc<WorkspaceManager>,
187 disc: Arc<DiscernPipeline>,
188 pool: Option<Arc<brahman_sidecar::SidecarPool>>,
189) -> anyhow::Result<()> {
190 loop {
191 let req: Request = match read_frame(&mut stream).await {
192 Ok(r) => r,
193 Err(shipote_protocol::ProtocolError::Closed) => return Ok(()),
194 Err(e) => return Err(e.into()),
195 };
196 let resp = dispatch(&mgr, &disc, &pool, req).await;
197 write_frame(&mut stream, &resp).await?;
198 }
199}
200
201async fn dispatch(
202 mgr: &Arc<WorkspaceManager>,
203 disc: &DiscernPipeline,
204 pool: &Option<Arc<brahman_sidecar::SidecarPool>>,
205 req: Request,
206) -> Response {
207 match req {
208 Request::Ping => Response::Pong,
209
210 Request::WorkspaceCreate { spec } => match mgr.create(spec).await {
211 Ok((id, warnings)) => Response::WorkspaceCreated { id, warnings },
212 Err(e) => Response::Error { message: format!("{e}") },
213 },
214
215 Request::WorkspaceList => {
216 let items = mgr
217 .list()
218 .await
219 .into_iter()
220 .map(|s| WorkspaceSummary {
221 id: s.id,
222 label: s.label,
223 commands: s.commands,
224 uptime_ms: s.uptime_ms,
225 })
226 .collect();
227 Response::WorkspaceList { items }
228 }
229
230 Request::WorkspaceStop { id, grace_ms } => {
231 match mgr
232 .stop_with_grace(id, std::time::Duration::from_millis(grace_ms))
233 .await
234 {
235 Ok(reaped) => Response::WorkspaceStopped { id, reaped },
236 Err(e) => Response::Error { message: format!("{e}") },
237 }
238 }
239
240 Request::Run { workspace, exec, argv, envp, restart_on_failure } => {
241 match mgr
242 .run_with_options(workspace, exec, argv, envp, restart_on_failure)
243 .await
244 {
245 Ok(s) => Response::RunStarted {
246 workspace,
247 command_id: s.id,
248 pid: s.pid,
249 },
250 Err(e) => Response::Error { message: format!("{e}") },
251 }
252 }
253
254 Request::PipelineRun { spec, tap, vars } => {
255 let vars_map: std::collections::HashMap<String, String> = vars.into_iter().collect();
256 let spec = match shipote_card::substitute_vars(&spec, &vars_map) {
257 Ok(s) => s,
258 Err(e) => return Response::Error { message: format!("template: {e}") },
259 };
260 let disc = DiscernPipeline::default_pipeline();
261 let inc = mgr.incarnator_handle();
262 let ws_label = mgr.workspace_label(spec.workspace).await.unwrap_or_default();
263 match shipote_core::pipeline::run_pipeline(
264 &spec,
265 &ws_label,
266 tap,
267 std::sync::Arc::new(disc),
268 inc,
269 Some(mgr.clone()),
270 )
271 .await
272 {
273 Ok(launch) => {
274 let pipeline_id = launch.pipeline;
275 announce_edges_to_broker(pool.as_deref(), &pipeline_id, &launch.edge_discernments);
276 let cmds = launch.command_pids;
277 mgr.register_pipeline_commands(spec.workspace, pipeline_id, cmds.clone()).await;
278 mgr.register_pipeline_supervisor(pipeline_id, spec.workspace, spec.clone(), tap).await;
279 let edges = launch.edge_discernments.into_iter().map(map_edge_to_info).collect();
280 Response::PipelineStarted {
281 pipeline: pipeline_id,
282 command_pids: cmds,
283 edges,
284 }
285 }
286 Err(e) => Response::Error { message: format!("{e}") },
287 }
288 }
289
290 Request::Discern { sample, hint_path } => {
291 let path_str = hint_path.as_ref().and_then(|p| p.to_str());
292 let hint = Hint {
293 path: path_str,
294 size_total: None,
295 };
296 match disc.discern(&sample, &hint) {
297 Some(d) => Response::Discernment {
298 ty: format!("{:?}", d.ty),
299 confidence: d.confidence,
300 mime: d.mime,
301 lens: d.lens,
302 },
303 None => Response::Error { message: "no discernment".into() },
304 }
305 }
306
307 Request::CommandList { workspace } => {
308 let items: Vec<ProtoCommandInfo> = mgr
309 .list_commands(workspace)
310 .await
311 .into_iter()
312 .map(|c| ProtoCommandInfo {
313 id: c.id,
314 label: c.label,
315 pid: c.pid,
316 alive: c.alive,
317 exit_status: c.exit_status,
318 log_bytes: c.log_bytes,
319 })
320 .collect();
321 Response::CommandList { items }
322 }
323
324 Request::CommandLogs { workspace, command, tail_bytes, stream } => {
325 let s = match stream.as_str() {
326 "stdout" => shipote_core::LogStream::Stdout,
327 "stderr" => shipote_core::LogStream::Stderr,
328 _ => shipote_core::LogStream::Both,
329 };
330 match mgr.get_command_logs(workspace, command, tail_bytes, s).await {
331 Some(bytes) => Response::CommandLogs { bytes },
332 None => Response::Error {
333 message: format!("no logs for command {command} in workspace {workspace}"),
334 },
335 }
336 }
337
338 Request::PipelineSave { name, spec } => {
339 mgr.save_pipeline(name.clone(), spec).await;
340 Response::PipelineSaved { name }
341 }
342
343 Request::PipelineSavedList => {
344 let names = mgr.list_saved_pipelines().await;
345 Response::PipelineSavedList { names }
346 }
347
348 Request::PipelineDrop { name } => {
349 let existed = mgr.drop_saved_pipeline(&name).await;
350 Response::PipelineDropped { name, existed }
351 }
352
353 Request::PipelineRunSaved { name, tap, vars } => match mgr.get_saved_pipeline(&name).await {
354 Some(spec) => {
355 let vars_map: std::collections::HashMap<String, String> = vars.into_iter().collect();
356 let spec = match shipote_card::substitute_vars(&spec, &vars_map) {
357 Ok(s) => s,
358 Err(e) => return Response::Error { message: format!("template: {e}") },
359 };
360 let disc = DiscernPipeline::default_pipeline();
361 let inc = mgr.incarnator_handle();
362 let ws_label = mgr.workspace_label(spec.workspace).await.unwrap_or_default();
363 match shipote_core::pipeline::run_pipeline(
364 &spec,
365 &ws_label,
366 tap,
367 std::sync::Arc::new(disc),
368 inc,
369 Some(mgr.clone()),
370 )
371 .await
372 {
373 Ok(launch) => {
374 let pipeline_id = launch.pipeline;
375 announce_edges_to_broker(pool.as_deref(), &pipeline_id, &launch.edge_discernments);
376 let cmds = launch.command_pids;
377 mgr.register_pipeline_commands(spec.workspace, pipeline_id, cmds.clone()).await;
378 mgr.register_pipeline_supervisor(pipeline_id, spec.workspace, spec.clone(), tap).await;
379 let edges = launch.edge_discernments.into_iter().map(map_edge_to_info).collect();
380 Response::PipelineStarted {
381 pipeline: pipeline_id,
382 command_pids: cmds,
383 edges,
384 }
385 }
386 Err(e) => Response::Error { message: format!("{e}") },
387 }
388 }
389 None => Response::Error {
390 message: format!("pipeline `{name}` no encontrado"),
391 },
392 },
393
394 Request::PipelineStop { pipeline, grace_ms } => {
395 let reaped = mgr
396 .stop_pipeline(pipeline, std::time::Duration::from_millis(grace_ms))
397 .await;
398 Response::PipelineStopped { pipeline, reaped }
399 }
400
401 Request::WorkspaceStats { workspace } => match mgr.workspace_stats(workspace).await {
402 Some(s) => Response::WorkspaceStats {
403 info: WorkspaceStatsInfo {
404 commands_alive: s.commands_alive,
405 commands_total: s.commands_total,
406 rss_bytes: s.rss_bytes,
407 rss_peak_bytes: s.rss_peak_bytes,
408 cpu_usec: s.cpu_usec,
409 cpu_percent: s.cpu_percent,
410 cpu_cores: s.cpu_cores,
411 source: s.source,
412 uptime_ms: s.uptime_ms,
413 },
414 },
415 None => Response::Error {
416 message: format!("workspace {workspace} not found"),
417 },
418 },
419
420 Request::WorkspaceQuota { workspace } => match mgr.workspace_quota(workspace).await {
421 Some(q) => Response::WorkspaceQuota {
422 info: QuotaReportInfo {
423 mem_limit: q.mem_limit,
424 nproc_limit: q.nproc_limit,
425 breaches: q.breaches,
426 },
427 },
428 None => Response::Error {
429 message: format!("workspace {workspace} not found"),
430 },
431 },
432
433 Request::FlowList => {
434 let items = mgr
435 .list_flow_pipelines()
436 .await
437 .into_iter()
438 .map(|(pipeline, sockets)| FlowInfo { pipeline, sockets })
439 .collect();
440 Response::FlowList { items }
441 }
442
443 Request::FlowDrop { pipeline } => {
444 let existed = mgr.drop_pipeline_flows(pipeline).await;
445 Response::FlowDropped { pipeline, existed }
446 }
447
448 Request::Capabilities => {
449 let c = mgr.incarnator().capabilities();
450 Response::Capabilities {
451 kernel_version: c.kernel_version,
452 user_ns: format!("{:?}", c.user_ns),
453 cgroup_v2: format!("{:?}", c.cgroup_v2),
454 cgroup_delegated: c.cgroup_delegated,
455 has_cap_sys_admin: c.has_cap_sys_admin,
456 }
457 }
458 }
459}
460
461fn map_edge_to_info(e: shipote_core::pipeline::EdgeDiscernment) -> EdgeDiscernmentInfo {
462 EdgeDiscernmentInfo {
463 from_label: e.from_label,
464 from_output: e.from_output,
465 to_label: e.to_label,
466 to_input: e.to_input,
467 ty: e.discernment.as_ref().map(|d| format!("{:?}", d.ty)),
468 mime: e.discernment.as_ref().and_then(|d| d.mime.clone()),
469 lens: e.discernment.as_ref().and_then(|d| d.lens.clone()),
470 confidence: e.discernment.as_ref().map(|d| d.confidence).unwrap_or(0.0),
471 flow_socket: e.flow_socket,
472 }
473}
474
475/// Por cada edge con TypeRef detectado, spawneamos una Card efímera en el
476/// SidecarPool que se anuncia al broker como producer del TypeRef
477/// enriquecido. Esto permite a otros explorers (broker-explorer, etc.)
478/// ver que shipote vio JSON/text/wasm/etc. saliendo de un pipeline.
479fn announce_edges_to_broker(
480 pool: Option<&brahman_sidecar::SidecarPool>,
481 pipeline: &ulid::Ulid,
482 edges: &[shipote_core::pipeline::EdgeDiscernment],
483) {
484 let Some(pool) = pool else { return };
485 for e in edges {
486 let Some(d) = &e.discernment else { continue };
487 let label = format!(
488 "shipote.flow.{}.{}.{}.{}",
489 short_ulid(pipeline),
490 e.from_label,
491 e.from_output,
492 type_label(&d.ty)
493 );
494 let mut card = Card::new(label);
495 card.kind = CardKind::Data;
496 card.lifecycle = Lifecycle::Oneshot;
497 card.payload = Payload::Virtual;
498 card.supervision = Supervision::OneShot;
499 card.flow = Flows {
500 input: Vec::new(),
501 output: vec![Flow {
502 name: e.from_output.clone(),
503 ty: d.ty.clone(),
504 pin_to: None,
505 }],
506 };
507 pool.spawn(card);
508 info!(pipeline = %pipeline, from = %e.from_label, ty = ?d.ty, "edge announced to broker");
509 }
510}
511
512fn short_ulid(u: &ulid::Ulid) -> String {
513 let s = u.to_string();
514 s[s.len() - 6..].to_string()
515}
516
517fn type_label(t: &TypeRef) -> String {
518 match t {
519 TypeRef::Primitive { name } => name.clone(),
520 TypeRef::Wit { package, name, .. } => format!("{package}.{name}"),
521 }
522}
523
524/// Card del daemon. La presentamos al broker así otras sesiones pueden
525/// descubrir que shipote está corriendo y, eventualmente, conectarse
526/// como consumidoras del flow `workspaces` (futuro: que la GUI o el
527/// broker-explorer los listen vía broker en lugar de socket directo).
528fn build_daemon_card(service_socket: &std::path::Path) -> Card {
529 let mut card = Card::new("shipote.daemon");
530 card.kind = CardKind::Ente;
531 card.lifecycle = Lifecycle::Daemon;
532 card.payload = Payload::Virtual; // el daemon ya está corriendo (no es PID 1 quien lo encarna)
533 card.supervision = Supervision::Delegate;
534 card.service_socket = Some(service_socket.to_path_buf());
535 card.flow = Flows {
536 input: Vec::new(),
537 output: vec![
538 Flow {
539 name: "workspaces".into(),
540 ty: TypeRef::Wit {
541 package: "shipote:admin".into(),
542 interface: None,
543 name: "workspace-list".into(),
544 },
545 pin_to: None,
546 },
547 Flow {
548 name: "discern".into(),
549 ty: TypeRef::Wit {
550 package: "shipote:admin".into(),
551 interface: None,
552 name: "discernment".into(),
553 },
554 pin_to: None,
555 },
556 ],
557 };
558 card
559}
560
561fn init_tracing() {
562 use tracing_subscriber::{fmt, EnvFilter};
563 let filter = EnvFilter::try_from_env("SHIPOTE_LOG").unwrap_or_else(|_| EnvFilter::new("info"));
564 fmt().with_env_filter(filter).init();
565}