mecha10_core/

prelude.rs

pub use crate::context::{
    Context, InfrastructureConfig, MinioConfig, MlflowConfig, PostgresConfig, Receiver, RedisConfig,
};

#[cfg(feature = "messaging")]
pub use crate::stream::ReceiverExt;

#[cfg(feature = "messaging")]
pub use crate::publish::ContextPublishExt;

#[cfg(feature = "messaging")]
pub use crate::rpc::{Request, Response, RpcExt};

#[cfg(feature = "messaging")]
pub use crate::rpc_streaming::{
    StreamError, StreamMessage, StreamPayload, StreamReceiver, StreamSender, StreamingRpcExt,
};

#[cfg(feature = "messaging")]
pub use crate::discovery::DiscoveryExt;
pub use crate::discovery::{NodeMetadata, NodeStatus};

pub use crate::schema::{
    CompatibilityCheck, CompatibilityMode, MessageSchema, MessageVersionInfo, SchemaEvolution, SchemaField,
    SchemaRegistryExt, SchemaVersioningExt,
};

pub use crate::auth::{AuthExt, Permission, Policy, Rule};

pub use crate::secrets::{
    EnvironmentBackend, RedisBackend, SecretBackend, SecretsConfig, SecretsManager, VaultBackend,
};

#[cfg(feature = "messaging")]
pub use crate::health::{
    HealthMonitor, HealthReport, HealthReporter, HealthReportingExt, NodeHealthInfo, SystemHealth,
};

#[cfg(feature = "messaging")]
pub use crate::behavior_interrupt::{
    BehaviorControl, BehaviorInterruptConfig, BehaviorInterruptTrigger, InterruptMode,
};

#[cfg(feature = "messaging")]
pub use crate::on_demand::subscribe_on_demand;
pub use crate::on_demand::OnDemandRequest;

pub use crate::metrics::{Counter, Gauge, Histogram, MetricsRegistry, NodeMetrics};

pub use crate::rate_limit::{MultiKeyRateLimiter, RateLimit, RateLimiter, RateLimiterManager};

pub use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState};

#[cfg(feature = "messaging")]
pub use crate::dead_letter::{DeadLetterConfig, DeadLetterQueue, FailedMessage};

#[cfg(feature = "schema-validation")]
pub use crate::schema_validation::{
    export_project_schema, get_project_schema, load_and_validate_project, validate_config_file,
    validate_node_config_file, validate_node_config_json, validate_project_config, validate_project_json,
};

pub use crate::dependency::{create_graph, DependencyGraph, ShutdownPlan, StartupPlan};

#[cfg(feature = "messaging")]
pub use crate::lifecycle::{
    get_timestamp_ms, topic_node_lifecycle, topic_node_status, topic_node_trigger, ModeChanged, ModeError, ModeRequest,
    NodeLifecycleAction, NodeLifecycleEvent, NodeLifecycleStatus, NodeStatusEvent, NodeTriggerEvent,
    TOPIC_MODE_CHANGED, TOPIC_MODE_ERROR, TOPIC_MODE_REQUEST,
};

pub use crate::topic_registry::{TopicInfo, TopicRegistry, TopologyStats};

pub use crate::topic_utils::{extract_category, format_elapsed_time, TopicMetadata, Topics};

pub use crate::state::{
    FilesystemStateManager, MemoryStateManager, StateManager, StateOptions, StateSnapshot, VersionedState,
};

pub use crate::recovery::{
    BackoffStrategy, DegradationMode, RecoveryAction, RecoveryAttempt, RecoveryHistory, RecoveryManager,
    RecoveryPolicy, RecoveryPolicyBuilder, RecoveryStats,
};

#[cfg(feature = "messaging")]
pub use crate::tracing_otel::{
    JaegerExporter, SpanBuilder, SpanData, TraceContext, TraceExporter, TracedMessage, TracingConfig, TracingExt,
    TracingHandle, ZipkinExporter,
};

pub use crate::profiling::{
    measure_latency, ExecutionStats, LatencyStats, MemoryStats, PerformanceMetrics, PerformanceReport, ProfilingExt,
    ThroughputStats, TimerGuard,
};

pub use crate::service::Service;
#[cfg(feature = "messaging")]
pub use crate::service_manager::ServiceManager;

pub use crate::node::{
    fmt_node_with_controller, run_node, HealthPriority, HealthReportingConfig, HealthStatusExt, Node, NodeImpl,
    NodeTag, NodeType, RateConfig,
};

pub use crate::config::{
    env_or_default, load_and_validate, load_config, load_config_from_env, substitute_env_vars, Validate,
};

pub use crate::model::{load_labels, load_model_config, CustomLabelsConfig, ModelConfig, PreprocessingConfig};

pub use crate::builder_setter;
pub use crate::enum_str_method;

pub use crate::messages::Message;

pub use crate::messages::Timestamped;

pub use crate::types::*;

pub use crate::error::{Mecha10Error, Result};

pub use async_trait::async_trait;

pub use tokio::{
    self, select,
    time::{interval, sleep, timeout, Duration, Instant},
};

pub use tokio::task::JoinHandle;

pub use serde::{Deserialize, Serialize};
pub use serde_json::{self, json, Value as JsonValue};

pub use schemars::JsonSchema;

pub use tracing::{debug, error, info, instrument, trace, warn, Level};

#[cfg(feature = "macros")]
pub use mecha10_macros::Node;

#[cfg(feature = "macros")]
pub use mecha10_macros::Config;

#[cfg(feature = "macros")]
pub use mecha10_macros::ConfigDefaults;

/// Shared implementation behind `now_micros`/`now_secs`/`now_millis`, which differ only in
/// which `Duration` unit they read off the epoch offset.
fn since_epoch() -> std::time::Duration {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .expect("Time went backwards")
}

/// Get current time in microseconds since UNIX epoch
pub fn now_micros() -> u64 {
    since_epoch().as_micros() as u64
}

/// Get current time in whole seconds since UNIX epoch
pub fn now_secs() -> u64 {
    since_epoch().as_secs()
}

/// Create a periodic interval from a frequency in Hz
///
/// This is a convenience helper that converts frequency (Hz) to a tokio interval.
/// Commonly used for periodic publishing in drivers and nodes.
///
/// # Arguments
///
/// * `hz` - Frequency in Hertz (cycles per second)
///
/// # Returns
///
/// A tokio::time::Interval that ticks at the specified frequency
///
/// # Example
///
/// Marked `no_run`: the `loop` below never breaks, so running it to completion
/// would hang the doctest forever.
/// ```rust,no_run
/// use mecha10_core::prelude::*;
///
/// # async fn example() {
/// // Instead of:
/// // let interval_duration = Duration::from_secs_f32(1.0 / 30.0);
/// // let mut interval = tokio::time::interval(interval_duration);
///
/// // Simply use:
/// let mut interval = interval_at_hz(30.0);
///
/// loop {
///     interval.tick().await;
///     // Publish at 30 Hz
/// }
/// # }
/// ```
pub fn interval_at_hz(hz: f32) -> tokio::time::Interval {
    let duration = std::time::Duration::from_secs_f32(1.0 / hz);
    tokio::time::interval(duration)
}

/// Initialize logging with sensible defaults
///
/// This function is safe to call multiple times - it will silently succeed if
/// a tracing subscriber is already set (e.g., when running bundled in the CLI).
///
/// The logging system will automatically include node context (node_id) in all
/// log messages when using the Context's logging methods or the `in_node_context!` macro.
///
/// Under the `messaging` feature, this also attaches
/// [`crate::log_publisher::RedisLogLayer`], the layer that ships captured
/// log events to Redis on `/system/log` once `crate::log_publisher::attach`
/// has been called with a live `Context` (done by `run_node`). It's
/// attached here - unconditionally, before any node-specific state exists -
/// rather than deferred, because `init_logging()` is what every node
/// binary's `main.rs` calls *first* (before `run_node` ever runs): `tracing`
/// only lets the first `try_init()` call in a process actually install a
/// subscriber, so the redis layer has to ride along with whichever call
/// wins that race, not assume it'll be `run_node`'s.
pub fn init_logging() {
    use tracing_subscriber::EnvFilter;

    static SENTRY_GUARD: std::sync::OnceLock<sentry::ClientInitGuard> = std::sync::OnceLock::new();
    if SENTRY_GUARD.get().is_none() && sentry::Hub::current().client().is_none() {
        let sentry_dsn = std::env::var("SENTRY_DSN_NODES").unwrap_or_else(|_| {
            "https://7ea967345cb382353a93ce31365e4790@o4506107376631808.ingest.us.sentry.io/4512334108229632"
                .to_string()
        });
        let mut sentry_opts = sentry::ClientOptions::default();
        sentry_opts.dsn = sentry_dsn.parse().ok();
        sentry_opts.release = sentry::release_name!();
        let _ = SENTRY_GUARD.set(sentry::init(sentry_opts));
    }

    let base = std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string());
    let curated = format!("{base},ort=warn,rtc_dtls=error,rtc_ice=error,rtc::peer_connection::handler=error");
    let env_filter = EnvFilter::try_new(&curated).unwrap_or_else(|_| {
        EnvFilter::new("info,ort=warn,rtc_dtls=error,rtc_ice=error,rtc::peer_connection::handler=error")
    });

    #[cfg(feature = "messaging")]
    {
        use tracing_subscriber::prelude::*;

        let _ = tracing_subscriber::registry()
            .with(env_filter)
            .with(tracing_subscriber::fmt::layer())
            .with(crate::log_publisher::RedisLogLayer::new())
            .try_init();
    }

    #[cfg(not(feature = "messaging"))]
    {
        let _ = tracing_subscriber::fmt().with_env_filter(env_filter).try_init();
    }
}

#[macro_export]
macro_rules! in_node_context {
    ($ctx:expr, $body:block) => {{
        let _span = tracing::info_span!("node", node_id = $ctx.node_id()).entered();
        $body
    }};
}

pub use in_node_context;

#[macro_export]
macro_rules! node_binary_main {
    ($node_crate:ident) => {
        #[tokio::main]
        async fn main() -> anyhow::Result<()> {
            $crate::prelude::init_logging();

            $node_crate::run().await.map_err(|e| anyhow::anyhow!(e))
        }
    };
}

pub use node_binary_main;

/// Get the current time in milliseconds since UNIX epoch
///
/// This is useful when microsecond precision is excessive (logs, metrics, etc.).
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let timestamp_ms = now_millis();
/// println!("Current time: {} ms", timestamp_ms);
/// ```
pub fn now_millis() -> u64 {
    since_epoch().as_millis() as u64
}

/// Calculate time elapsed in microseconds since a given timestamp
///
/// # Arguments
///
/// * `timestamp_micros` - A timestamp in microseconds (from `now_micros()`)
///
/// # Returns
///
/// Elapsed time in microseconds
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let start = now_micros();
/// // ... do work ...
/// let elapsed = elapsed_micros(start);
/// println!("Operation took {} μs", elapsed);
/// ```
pub fn elapsed_micros(timestamp_micros: u64) -> u64 {
    now_micros().saturating_sub(timestamp_micros)
}

/// Convert a microsecond timestamp to seconds (as f64)
///
/// Useful for displaying timestamps in human-readable format.
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let timestamp = now_micros();
/// let seconds = timestamp_as_secs(timestamp);
/// println!("Time: {:.6} seconds", seconds);
/// ```
pub fn timestamp_as_secs(timestamp_micros: u64) -> f64 {
    timestamp_micros as f64 / 1_000_000.0
}

/// Create a Duration from a frequency in Hz
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let duration = duration_from_hz(30.0);  // 33.33ms for 30Hz
/// ```
pub fn duration_from_hz(hz: f32) -> Duration {
    Duration::from_secs_f32(1.0 / hz)
}

/// Normalize an angle in radians to the range [-π, π]
///
/// This is essential for angle calculations to avoid unwanted wraparound effects.
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let angle = normalize_angle(7.0);  // Returns ~0.717 (7.0 - 2π)
/// let angle2 = normalize_angle(-4.0); // Returns ~2.283 (-4.0 + 2π)
/// ```
pub fn normalize_angle(angle: f32) -> f32 {
    use std::f32::consts::PI;
    let mut normalized = angle;
    while normalized > PI {
        normalized -= 2.0 * PI;
    }
    while normalized < -PI {
        normalized += 2.0 * PI;
    }
    normalized
}

/// Normalize an angle in degrees to the range [-180, 180]
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let angle = normalize_angle_degrees(270.0);  // Returns -90.0
/// ```
pub fn normalize_angle_degrees(angle: f32) -> f32 {
    let mut normalized = angle;
    while normalized > 180.0 {
        normalized -= 360.0;
    }
    while normalized < -180.0 {
        normalized += 360.0;
    }
    normalized
}

/// Calculate the shortest angular difference between two angles in radians
///
/// Returns the angular distance in the range [-π, π].
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let diff = angle_difference(0.1, 6.2);  // Shortest path wrapping around
/// ```
pub fn angle_difference(from: f32, to: f32) -> f32 {
    normalize_angle(to - from)
}

/// Convert degrees to radians
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let radians = degrees_to_radians(180.0);  // Returns π
/// ```
pub fn degrees_to_radians(degrees: f32) -> f32 {
    degrees * std::f32::consts::PI / 180.0
}

/// Convert radians to degrees
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let degrees = radians_to_degrees(std::f32::consts::PI);  // Returns 180.0
/// ```
pub fn radians_to_degrees(radians: f32) -> f32 {
    radians * 180.0 / std::f32::consts::PI
}

/// Safely divide two numbers, returning a default value if divisor is zero
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let result = safe_divide(10.0, 2.0, 0.0);   // Returns 5.0
/// let result = safe_divide(10.0, 0.0, 99.0);  // Returns 99.0 (default)
/// ```
pub fn safe_divide(numerator: f32, denominator: f32, default: f32) -> f32 {
    if denominator.abs() < f32::EPSILON {
        default
    } else {
        numerator / denominator
    }
}

/// Clamp a value between a minimum and maximum
///
/// This is available in std but this wrapper makes it more discoverable.
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let value = clamp(15.0, 0.0, 10.0);  // Returns 10.0
/// ```
pub fn clamp<T: PartialOrd>(value: T, min: T, max: T) -> T {
    if value < min {
        min
    } else if value > max {
        max
    } else {
        value
    }
}

/// Linear interpolation between two values
///
/// # Arguments
///
/// * `a` - Start value
/// * `b` - End value
/// * `t` - Interpolation factor [0.0, 1.0]
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// let mid = lerp(0.0, 10.0, 0.5);  // Returns 5.0
/// ```
pub fn lerp(a: f32, b: f32, t: f32) -> f32 {
    a + (b - a) * t
}

pub use std::collections::{HashMap, HashSet};

pub use std::sync::{Arc, Mutex};

pub use std::path::{Path, PathBuf};

pub type Shared<T> = Arc<T>;
pub type SharedMut<T> = Arc<Mutex<T>>;

#[cfg(test)]
mod sentry_options_tests {
    // Regression test for the E0639 compile error: `sentry::ClientOptions` is
    // `#[non_exhaustive]`, so it cannot be built with a struct-literal +
    // `..Default::default()`. Construction must go through
    // `ClientOptions::default()` followed by field assignment, as done in
    // `init_logging`. If a future edit reverts to the struct-literal form,
    // this module (and `init_logging` itself) will fail to compile.
    #[test]
    fn client_options_builds_via_default_and_field_assignment() {
        let dsn = "https://public@sentry.example.com/1";

        let mut opts = sentry::ClientOptions::default();
        opts.dsn = dsn.parse().ok();
        opts.release = sentry::release_name!();

        assert!(opts.dsn.is_some());
    }

    #[test]
    fn invalid_dsn_parses_to_none_without_panicking() {
        let mut opts = sentry::ClientOptions::default();
        opts.dsn = "not-a-valid-dsn".parse().ok();

        assert!(opts.dsn.is_none());
    }
}