mecha10_core/

lib.rs

//! Mecha10 Core Runtime
//!
//! This crate provides the foundational runtime for the Mecha10 framework,
//! including node lifecycle management, health monitoring, and orchestration.
//!
//! # Features
//!
//! - Node lifecycle management (start, stop, restart)
//! - Health monitoring and automatic recovery
//! - Process management for node execution
//! - Integration with config and messaging systems
//! - Graceful shutdown handling
//! - Type-safe topics system
//! - Comprehensive error handling
//! - Shared message types
//!
//! # Quick Start
//!
//! ```rust
//! use mecha10_core::prelude::*;
//! use mecha10_core::topics::{perception, sensor};
//!
//! #[derive(Debug)]
//! struct MyNode {
//!     config: MyConfig,
//! }
//!
//! #[derive(Debug, Deserialize)]
//! struct MyConfig {
//!     threshold: f32,
//! }
//!
//! #[async_trait]
//! impl NodeImpl for MyNode {
//!     type Config = MyConfig;
//!
//!     async fn init(config: Self::Config) -> Result<Self> {
//!         Ok(Self { config })
//!     }
//!
//!     async fn run(&mut self, ctx: &Context) -> Result<()> {
//!         let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
//!
//!         while let Some(image) = images.recv().await {
//!             // Process image into detections, then publish the result.
//!             let result = Vec::new();
//!             ctx.publish_to(perception::DETECTIONS, &result).await?;
//!         }
//!
//!         Ok(())
//!     }
//! }
//! ```

pub mod types;

pub mod wire_identifier;

pub mod messages;

pub mod sensor;

pub mod actuator;

pub mod teleop;

pub mod on_demand;

pub mod topics;

pub mod topic_utils;

pub mod error;

pub mod fs_utils;

pub mod rwlock_collections;

pub mod context;

#[cfg(feature = "messaging")]
pub mod stream;

#[cfg(feature = "messaging")]
pub mod publish;

pub mod node;

pub mod node_registry;

pub mod config;

pub mod config_watcher;

pub mod model;

pub mod macros;

#[cfg(feature = "messaging")]
pub mod rpc;

#[cfg(feature = "messaging")]
pub mod rpc_breaker;

#[cfg(feature = "messaging")]
pub mod rpc_streaming;

pub mod discovery;

pub mod schema;

pub mod auth;

mod redis_list_util;

pub mod secrets;

#[cfg(feature = "messaging")]
pub mod health;

#[cfg(feature = "messaging")]
pub mod log_publisher;

pub mod metrics;

pub mod rate_limit;

pub mod circuit_breaker;

#[cfg(feature = "messaging")]
pub mod dead_letter;

#[cfg(feature = "schema-validation")]
pub mod schema_validation;

#[cfg(feature = "hal-compat")]
pub mod hal_compat;

pub mod dependency;

pub mod kinematics;

pub mod geometry;

#[cfg(feature = "messaging")]
pub mod lifecycle;

pub mod topic_registry;

pub mod state;

pub mod recovery;

#[cfg(feature = "messaging")]
pub mod tracing_otel;

pub mod profiling;

pub mod priority_queue;

pub mod service;
#[cfg(feature = "messaging")]
pub mod service_manager;

pub mod aggregated;

#[cfg(feature = "messaging")]
pub mod recorder;

#[cfg(feature = "messaging")]
pub mod redis_bridge;

#[cfg(feature = "messaging")]
pub mod behavior_interrupt;

#[cfg(feature = "messaging")]
pub mod command_relay;

pub mod relay_topics;

pub mod node_config_base;

pub mod ollama;

pub mod prelude;

pub use context::{Context, Receiver};
pub use error::{Mecha10Error, Result};
pub use messages::Message;
pub use types::*;

// Re-exported so `register_node!`'s `$crate::inventory::submit!` resolves from any crate that
// invokes it (`$crate` in a `#[macro_export]` macro always points back at this crate, regardless
// of where the macro is called from).
pub use inventory;

use std::collections::HashMap;
use std::process::Stdio;
use std::sync::Arc;
use thiserror::Error;
use tokio::process::Child;
use tokio::sync::RwLock;
use tokio::time::{interval, Duration};

use crate::rwlock_collections::RwLockMapExt;

#[derive(Debug, Error)]
pub enum RuntimeError {
    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),

    #[error("Node error: {0}")]
    Node(String),

    #[error("Configuration error: {0}")]
    Config(String),

    #[error("Process error: {0}")]
    Process(String),

    #[error("Health check failed: {0}")]
    HealthCheck(String),
}

pub type RuntimeResult<T> = std::result::Result<T, RuntimeError>;

struct ManagedNode {
    config: NodeConfig,
    state: NodeState,
    process: Option<Child>,
    restart_count: u32,
    last_health_check: std::time::Instant,
}

impl ManagedNode {
    fn new(config: NodeConfig) -> Self {
        Self {
            config,
            state: NodeState::Stopped,
            process: None,
            restart_count: 0,
            last_health_check: std::time::Instant::now(),
        }
    }

    async fn start(&mut self) -> RuntimeResult<()> {
        if self.state == NodeState::Running {
            return Ok(());
        }

        self.state = NodeState::Starting;

        let mut cmd = tokio::process::Command::new(&self.config.executable);
        cmd.args(&self.config.args);
        cmd.envs(&self.config.env);
        cmd.stdout(Stdio::piped());
        cmd.stderr(Stdio::piped());

        let child = cmd
            .spawn()
            .map_err(|e| RuntimeError::Process(format!("Failed to spawn {}: {}", self.config.id, e)))?;

        self.process = Some(child);
        self.state = NodeState::Running;

        tracing::info!("Started node: {}", self.config.id);

        Ok(())
    }

    async fn stop(&mut self) -> RuntimeResult<()> {
        if let Some(mut child) = self.process.take() {
            self.state = NodeState::Stopping;

            #[cfg(unix)]
            {
                use nix::sys::signal::{self, Signal};
                use nix::unistd::Pid;

                if let Some(pid) = child.id() {
                    let _ = signal::kill(Pid::from_raw(pid as i32), Signal::SIGTERM);
                }
            }

            match tokio::time::timeout(Duration::from_secs(5), child.wait()).await {
                Ok(Ok(status)) => {
                    tracing::info!("Node {} stopped: {:?}", self.config.id, status);
                }
                Ok(Err(e)) => {
                    tracing::error!("Error waiting for node {}: {}", self.config.id, e);
                }
                Err(_) => {
                    tracing::warn!("Node {} did not stop gracefully, killing", self.config.id);
                    let _ = child.kill().await;
                }
            }

            self.state = NodeState::Stopped;
        }

        Ok(())
    }

    async fn check_health(&mut self) -> RuntimeResult<bool> {
        if let Some(child) = &mut self.process {
            match child.try_wait() {
                Ok(Some(status)) => {
                    tracing::warn!("Node {} exited: {:?}", self.config.id, status);
                    self.state = NodeState::Crashed;
                    self.process = None;
                    return Ok(false);
                }
                Ok(None) => {
                    self.last_health_check = std::time::Instant::now();
                    return Ok(true);
                }
                Err(e) => {
                    tracing::error!("Error checking node {}: {}", self.config.id, e);
                    self.state = NodeState::Unhealthy;
                    return Ok(false);
                }
            }
        }

        Ok(false)
    }

    fn should_restart(&self) -> bool {
        self.config.auto_restart
            && (self.state == NodeState::Crashed || self.state == NodeState::Unhealthy)
            && self.restart_count < 3
    }
}

pub struct Runtime {
    #[allow(dead_code)]
    robot_id: String,

    nodes: Arc<RwLock<HashMap<String, ManagedNode>>>,

    running: Arc<RwLock<bool>>,
}

impl Runtime {
    /// Create a new runtime instance
    pub async fn new(robot_id: &str) -> RuntimeResult<Self> {
        Ok(Self {
            robot_id: robot_id.to_string(),
            nodes: Arc::new(RwLock::new(HashMap::new())),
            running: Arc::new(RwLock::new(false)),
        })
    }

    /// Start a node
    pub async fn start_node(&mut self, config: NodeConfig) -> RuntimeResult<()> {
        let node_id = config.id.clone();

        let mut nodes = self.nodes.write().await;

        if nodes.contains_key(&node_id) {
            return Err(RuntimeError::Node(format!("Node {} already exists", node_id)));
        }

        let mut node = ManagedNode::new(config);
        node.start().await?;

        nodes.insert(node_id, node);

        Ok(())
    }

    /// Stop a node
    pub async fn stop_node(&mut self, node_id: &str) -> RuntimeResult<()> {
        let mut nodes = self.nodes.write().await;

        if let Some(node) = nodes.get_mut(node_id) {
            node.stop().await?;
            nodes.remove(node_id);
            Ok(())
        } else {
            Err(RuntimeError::Node(format!("Node {} not found", node_id)))
        }
    }

    /// Get node state
    pub async fn node_state(&self, node_id: &str) -> Option<NodeState> {
        let nodes = self.nodes.read().await;
        nodes.get(node_id).map(|n| n.state.clone())
    }

    /// List all nodes
    pub async fn list_nodes(&self) -> Vec<String> {
        self.nodes.cloned_keys().await
    }

    /// Run the runtime (blocks until shutdown)
    pub async fn run(&mut self) -> RuntimeResult<()> {
        *self.running.write().await = true;

        let nodes = Arc::clone(&self.nodes);
        let running = Arc::clone(&self.running);

        let health_check_task = tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(5));

            while *running.read().await {
                interval.tick().await;

                let mut nodes = nodes.write().await;
                let node_ids: Vec<String> = nodes.keys().cloned().collect();

                for node_id in node_ids {
                    if let Some(node) = nodes.get_mut(&node_id) {
                        if node.state == NodeState::Running {
                            match node.check_health().await {
                                Ok(true) => {}
                                Ok(false) => {
                                    if node.should_restart() {
                                        tracing::info!("Auto-restarting node: {}", node_id);
                                        let _ = node.start().await;
                                        node.restart_count += 1;
                                    }
                                }
                                Err(e) => {
                                    tracing::error!("Health check error for {}: {}", node_id, e);
                                }
                            }
                        }
                    }
                }
            }
        });

        tokio::signal::ctrl_c()
            .await
            .map_err(|e| RuntimeError::Process(format!("Error waiting for shutdown: {}", e)))?;

        tracing::info!("Shutting down runtime...");
        *self.running.write().await = false;

        health_check_task.abort();

        let mut nodes = self.nodes.write().await;
        for (node_id, node) in nodes.iter_mut() {
            tracing::info!("Stopping node: {}", node_id);
            let _ = node.stop().await;
        }

        Ok(())
    }

    /// Shutdown the runtime gracefully
    pub async fn shutdown(&mut self) -> RuntimeResult<()> {
        *self.running.write().await = false;

        let mut nodes = self.nodes.write().await;
        for (node_id, node) in nodes.iter_mut() {
            tracing::info!("Stopping node: {}", node_id);
            let _ = node.stop().await;
        }

        Ok(())
    }
}