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::prelude::*;
//!
//! #[derive(Node)]
//! #[subscribe(topics::sensor::CAMERA_RGB)]
//! #[publish(topics::perception::DETECTIONS)]
//! struct MyNode {
//!     config: MyConfig,
//! }
//!
//! #[derive(Config)]
//! struct MyConfig {
//!     #[validate(range = "0.0..=1.0")]
//!     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(topics::sensor::CAMERA_RGB).await?;
//!
//!         while let Some(image) = images.recv().await {
//!             // Process image
//!             ctx.publish(&result).await?;
//!         }
//!
//!         Ok(())
//!     }
//! }
//! ```

// ============================================================================
// Module Exports
// ============================================================================

// Core type system (all shared types)
pub mod types;

// Shared message types (sensor, actuator, perception, navigation, system)
pub mod messages;

// Shared sensor message types (camera, lidar, imu, odometry)
pub mod sensor;

// Shared actuator message types (motor commands, status)
pub mod actuator;

// Teleoperation message types (input commands, status)
pub mod teleop;

// Type-safe topics system
pub mod topics;

// Topic utilities (categorization, formatting, metadata)
pub mod topic_utils;

// Structured error handling
pub mod error;

// Node execution context
pub mod context;

// Stream combinators for declarative message processing
pub mod stream;

// Conditional publishing
pub mod publish;

// Node trait and lifecycle management
pub mod node;

// Configuration loading and validation
pub mod config;

// Configuration hot-reload support
pub mod config_watcher;

// Model configuration utilities for AI nodes
pub mod model;

// Convenience macros for reducing boilerplate
pub mod macros;

// Request/Response (RPC) pattern over pub/sub
pub mod rpc;

// RPC with circuit breaker integration
pub mod rpc_breaker;

// Streaming RPC for long-running operations
pub mod rpc_streaming;

// Service discovery for node registration and discovery
pub mod discovery;

// Schema registry for versioned message schemas
pub mod schema;

// Authorization and access control
pub mod auth;

// Secrets management
pub mod secrets;

// Health monitoring and aggregation
pub mod health;

// Metrics collection
pub mod metrics;

// Rate limiting
pub mod rate_limit;

// Circuit breaker for fault tolerance
pub mod circuit_breaker;

// Dead letter queue for failed message handling
pub mod dead_letter;

// JSON Schema validation for configuration files
pub mod schema_validation;

// Dependency resolution for node ordering
pub mod dependency;

// Lifecycle event schemas for mode-based management
pub mod lifecycle;

// Topic registry for runtime introspection
pub mod topic_registry;

// Persistent state storage abstraction
pub mod state;

// Graceful degradation and recovery policies
pub mod recovery;

// Distributed tracing
pub mod tracing_otel;

// Performance profiling helpers
pub mod profiling;

// Priority queue for quality-of-service
pub mod priority_queue;

// Service pattern for infrastructure components
pub mod service;
pub mod service_manager;

// Aggregated topics for multi-topic subscriptions
pub mod aggregated;

// Bag file recording and replay (ROS bags-like)
pub mod recorder;

// Redis bridge for local ↔ control plane message forwarding
pub mod redis_bridge;

// Behavior interrupt trigger for pausing autonomous behaviors
pub mod behavior_interrupt;

// Comprehensive prelude (single import for everything)
pub mod prelude;

// Re-export commonly used items at crate root
pub use context::{Context, Receiver};
pub use error::{Mecha10Error, Result};
pub use messages::Message;
pub use types::*;

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};

/// Core runtime errors
#[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),
}

/// Result type for runtime operations
pub type RuntimeResult<T> = std::result::Result<T, RuntimeError>;

/// Managed node instance
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;

            // Try graceful shutdown first
            #[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);
                }
            }

            // Wait for process to exit (with timeout)
            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(_) => {
                    // Timeout - force kill
                    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 {
            // Check if process is still alive
            match child.try_wait() {
                Ok(Some(status)) => {
                    // Process exited
                    tracing::warn!("Node {} exited: {:?}", self.config.id, status);
                    self.state = NodeState::Crashed;
                    self.process = None;
                    return Ok(false);
                }
                Ok(None) => {
                    // Process still running
                    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
    }
}

/// Runtime orchestrator for managing nodes
pub struct Runtime {
    /// Robot/instance identifier
    #[allow(dead_code)]
    robot_id: String,

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

    /// Running state
    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)))
        }
    }

    /// Restart a node
    pub async fn restart_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?;
            node.start().await?;
            node.restart_count += 1;
            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> {
        let nodes = self.nodes.read().await;
        nodes.keys().cloned().collect()
    }

    /// 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);

        // Health check loop
        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) => {
                                    // Node healthy
                                }
                                Ok(false) => {
                                    // Node unhealthy or crashed
                                    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);
                                }
                            }
                        }
                    }
                }
            }
        });

        // Wait for shutdown signal
        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();

        // Stop all nodes
        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(())
    }
}