horus_core

New to HORUS? This is the API reference - detailed documentation for experienced users. If you're just getting started, check out:

• Quick Start - Build your first application
• node! Macro - Write nodes with minimal code

The core runtime crate for the HORUS robotics framework. Provides the fundamental building blocks for creating distributed real-time robotics systems.

use horus::prelude::*;

Node

The fundamental trait for all computation units in HORUS.

pub trait Node: Send {
    // Required
    fn tick(&mut self);

    // Name (defaults to struct type name, e.g. `MotorController`)
    fn name(&self) -> &str { /* derived from type name */ }

    // Optional lifecycle
    fn init(&mut self) -> Result<()> { Ok(()) }
    fn shutdown(&mut self) -> Result<()> { Ok(()) }
    fn on_error(&mut self, _error: &str) {}  // no-op; `#[doc(hidden)]`, recovery hook only

    // Safety hooks
    fn is_safe_state(&self) -> bool { true }
    fn enter_safe_state(&mut self) {}

    // Runtime parameters (declared but not yet wired by the scheduler)
    fn on_parameter_change(
        &mut self,
        key: &str,
        old_value: Option<&serde_json::Value>,
        new_value: &serde_json::Value,
    ) -> Result<(), String> { Ok(()) }
}

That is the whole trait. There are no publishers()/subscribers() methods — topic associations are auto-detected — and no rate_hz(), is_healthy(), checkpoint_state(), restore_state() or supports_checkpointing() hooks.

error[E0107]: type alias takes 1 generic argument but 2 generic arguments were supplied

fn on_parameter_change(
    &mut self,
    key: &str,
    old_value: Option<&serde_json::Value>,
    new_value: &serde_json::Value,
) -> std::result::Result<(), String> {
    Ok(())
}

Required Methods

tick

fn tick(&mut self)

Called repeatedly by the scheduler. This is the main execution loop for the node.

Example:

fn tick(&mut self) {
    // Read sensor
    let value = self.sensor.read();

    // Publish data
    self.publisher.send(value);

    // Log using hlog! macro
    hlog!(debug, "Published: {}", value);
}

Optional Methods

init

fn init(&mut self) -> Result<()>

Called once before the first tick. Use for initialization that may fail.

Returns: Result<()> - Ok on success, Err on failure

Default: Returns Ok(())

shutdown

fn shutdown(&mut self) -> Result<()>

Called when the scheduler is stopping. Use for cleanup.

Returns: Result<()> - Ok on success, Err on failure

Default: Returns Ok(())

on_error

fn on_error(&mut self, _error: &str)

An error-recovery hook, called by the scheduler after a tick fails. Hidden from the generated rustdoc (#[doc(hidden)]) because it is not part of the reporting path.

Parameters:

• _error - Error message string

Default: No-op. Every caller — all four executors and the main-thread path — runs record_tick_failure first, which already logs the failure at error level to the console and to the buffer horus log reads. Override this only to add recovery logic, not to report the error.

name

fn name(&self) -> &str

The node's name, which must be unique within a scheduler. String literals and &self.name both work.

Returns: &str

Default: The struct's type name, e.g. MotorController

is_safe_state

fn is_safe_state(&self) -> bool

Reports whether the node is currently in a safe state. Queried by the safety monitor.

Returns: bool

Default: Returns true

enter_safe_state

fn enter_safe_state(&mut self)

Transitions the node to its safe state. Called for emergency stop — e.g. zero the motor outputs and drop the enable line.

Default: No-op

on_parameter_change

fn on_parameter_change(
    &mut self,
    key: &str,
    old_value: Option<&serde_json::Value>,
    new_value: &serde_json::Value,
) -> Result<(), String>

Intended to be called when a runtime parameter changes, letting the node inspect old/new values and return Err to reject the change.

Not yet wired: the scheduler does not currently call this method. Attaching RuntimeParams via the scheduler's .with_params() builder does not route RuntimeParams::set() to nodes, and there is no automatic rollback on reject — overriding this today has no effect. To react to parameter changes now, register a callback with RuntimeParams::on_change().

Default: Accepts all changes silently

Rate and topic association

Neither is a trait method:

• Per-node rate is set on the scheduler's node builder: scheduler.add(node).rate(1000_u64.hz()).build()?.
• Topic associations are automatic — they are detected via the topic/node registry when Topic::new() is called during a tick. There is nothing to declare.

Example Implementation

use horus::prelude::*;

struct TemperatureSensor {
    publisher: Topic<f32>,
    sample_count: u64,
}

impl TemperatureSensor {
    pub fn new() -> Result<Self> {
        Ok(Self {
            publisher: Topic::new("temperature")?,
            sample_count: 0,
        })
    }
}

impl Node for TemperatureSensor {
    fn name(&self) -> &str {
        "temperature_sensor"
    }

    fn init(&mut self) -> Result<()> {
        hlog!(info, "Temperature sensor initialized");
        Ok(())
    }

    fn tick(&mut self) {
        // Simulate reading temperature
        let temp = 20.0 + (self.sample_count as f32 * 0.1).sin() * 5.0;

        self.publisher.send(temp);
        self.sample_count += 1;
    }

    fn shutdown(&mut self) -> Result<()> {
        hlog!(info, "Sensor shutdown after {} samples", self.sample_count);
        Ok(())
    }
}

Topic

Unified typed pub/sub channel with automatic backend selection. Topic auto-selects the optimal IPC backend based on message type and usage patterns.

pub struct Topic<T> { /* fields omitted */ }

Type Parameters

• T - Message type. Must implement: Clone + Send + Sync + Serialize + DeserializeOwned + 'static

Backend Selection

Topic automatically selects the backend for your topology — you never need to choose one manually, and the backend types are not public API.

Every topic is shared-memory backed (a backing file is created and the creator PID stamped when the header is initialized), so every backend is cross-process; a consumer in another process that joins later reads published data immediately. The choice is a function of producer/consumer counts and whether the message type is POD:

Constructors

new

pub fn new(name: impl Into<String>) -> Result<Self>

Creates a new Topic with automatic backend selection and auto-sized capacity. Capacity is auto-calculated based on message size (one 4KB page of slots, clamped to 16-1024).

Parameters:

• name - Topic name or endpoint string

Returns: Result<Topic<T>>

Example:

let topic: Topic<f32> = Topic::new("sensor.temperature")?;

with_capacity

pub fn with_capacity(name: &str, capacity: u32, slot_size: Option<usize>) -> Result<Self>

Creates a Topic with explicit buffer capacity and optional slot size.

Parameters:

• name - Topic name
• capacity - Ring buffer capacity (rounded up to next power of 2)
• slot_size - Optional slot size in bytes (None uses default)

Returns: Result<Topic<T>>

Example:

let topic: Topic<Image> = Topic::with_capacity("camera.image", 16, None)?;

Methods

send

pub fn send(&self, msg: T)

Sends a message to the topic. Non-blocking, infallible. Uses bounded retry with "keep last" semantics — if the buffer is full after retry, the message is dropped.

When runtime debug logging is enabled (via the TUI monitor), metrics and message summaries are automatically recorded. Otherwise, zero-overhead.

Parameters:

• msg - Message to send

Example:

self.publisher.send(42.0);

try_send

pub fn try_send(&self, msg: T) -> Result<(), T>

Tries to send a message, returning it on failure for explicit retry. Unlike send(), does not retry — fails immediately if the buffer is full.

Parameters:

• msg - Message to send

Returns: Result<(), T> - Ok(()) on success, Err(msg) returns the message on failure

Example:

match topic.try_send(data) {
    Ok(()) => { /* sent */ }
    Err(returned) => { /* buffer full, try again later */ }
}

recv

pub fn recv(&self) -> Option<T>

Receives the next message from the topic. Non-blocking.

Returns: Option<T> - Some(message) if available, None otherwise

Example:

if let Some(value) = self.subscriber.recv() {
    // Process value
}

try_recv

pub fn try_recv(&self) -> Option<T>

Internal — prefer recv(). This is #[doc(hidden)] in the source and marked for internal and test use.

It does bypass the logging path, and it omits one thing recv() does: the check for a migration epoch change on an empty receive. recv() runs that check every time it returns None; try_recv() only reaches the amortized check inside the dispatch function, which a subscriber that never receives anything never triggers — so a try_recv-only subscriber can silently never see a cross-process publisher it should have migrated to.

It does still auto-register the caller as a subscriber — it calls register_sub on every call, exactly as recv() does — so a try_recv-only node appears normally in the topic and node registries.

Returns: Option<T> - Some(message) if available, None otherwise

read_latest

pub fn read_latest(&self) -> Option<T>
where
    T: Copy

Returns the most recent message without advancing the consumer position. Calling it multiple times returns the same message until a new one is published. Requires T: Copy for safety with multi-consumer backends.

Returns: Option<T> - The latest published message, or None if no messages have been published

Example:

// Read the latest transform without consuming it
if let Some(transform) = tf_topic.read_latest() {
    apply_transform(transform);
}

has_message

pub fn has_message(&self) -> bool

Checks if messages are available without consuming them.

Returns: bool

pending_count

pub fn pending_count(&self) -> u64

Returns the number of pending (unconsumed) messages in the buffer.

Returns: u64

Runtime Debug Logging

Debug logging is toggled at runtime from the TUI monitor by selecting a topic and pressing Enter. No code changes required.

When enabled, send() and recv() record timing, message summaries (if T: LogSummary), and IPC latency to the log buffer. When LogSummary is not implemented, metadata-only entries are logged.

horus monitor --tui    # Navigate to Topics tab, press Enter on a topic

name

pub fn name(&self) -> &str

Returns the topic name.

Returns: &str

metrics

pub fn metrics(&self) -> TopicMetrics

Returns current metrics snapshot (messages sent/received, send/recv failure counts).

Returns: TopicMetrics

TopicDescriptor and topics! Macro

Type-safe topic descriptors for compile-time checked topic names and message types.

use horus::topics;   // not in the prelude

/// Define type-safe topic descriptors
topics! {
    SENSOR_DATA: f64 = "sensor_data",
    MOTOR_CMD: MotorCommand = "motor_cmd",
}

// Use the descriptor's name to create a topic
let pub_topic: Topic<f64> = Topic::new(SENSOR_DATA.name())?;
let sub_topic: Topic<f64> = Topic::new(SENSOR_DATA.name())?;

TopicDescriptor

pub struct TopicDescriptor<T> {
    name: &'static str,
    _marker: PhantomData<T>,
}

impl<T> TopicDescriptor<T> {
    pub const fn new(name: &'static str) -> Self;
    pub const fn name(&self) -> &str;
}

Scheduler

Central orchestrator for node execution with priority-based scheduling.

pub struct Scheduler { /* fields omitted */ }

Constructors

new

pub fn new() -> Self

Creates a lightweight scheduler with default settings (100 Hz tick rate). Detects runtime capabilities (~30-100us) but does not apply any OS-level features. Use builder methods to opt in to features.

Returns: Scheduler

Example:

let mut scheduler = Scheduler::new();
scheduler.add(my_node).order(0).build()?;
scheduler.run()?;

Builder Methods

Scheduler::new() returns a Scheduler that is configured fluently: each method below takes self and returns Self, so chain them to set global behavior before adding nodes.

tick_rate

pub fn tick_rate(self, freq: Frequency) -> Self

Sets the global tick rate.

Example:

let scheduler = Scheduler::new().tick_rate(1000_u64.hz()); // 1 kHz

name

pub fn name(self, name: &str) -> Self

Sets the scheduler name for logging/monitoring. Defaults to "Scheduler", overridable via the HORUS_NODE_NAME environment variable.

deterministic

pub fn deterministic(self, enabled: bool) -> Self

Runs all nodes sequentially on the main thread in .order() order (ties broken by registration order) — no thread pools, no watcher threads, no executors in registration order — no thread pools, no watcher threads, no executors. Intended for simulation and testing with tick_once().

prefer_rt

pub fn prefer_rt(self) -> Self

Opts in to OS-level RT features (SCHED_FIFO, mlockall). Features that cannot be applied are recorded as degradations, not errors — this never panics.

Example:

let scheduler = Scheduler::new().prefer_rt();

require_rt

pub fn require_rt(self) -> Self

Same features as prefer_rt(), but fails loudly: panics if the system has neither SCHED_FIFO nor mlockall support.

Example:

let scheduler = Scheduler::new().require_rt();

cores

pub fn cores(self, cpu_ids: &[usize]) -> Self

Sets CPU affinity for the scheduler. This is a separate builder from the RT opt-in.

Example:

let scheduler = Scheduler::new().prefer_rt().cores(&[2, 3]);

blackbox

pub fn blackbox(self, size_mb: usize) -> Self

Enables the BlackBox flight recorder with the given buffer size in megabytes.

Example:

let scheduler = Scheduler::new().blackbox(16); // 16 MB flight recorder

watchdog

pub fn watchdog(self, timeout: Duration) -> Self

Enables the watchdog, which detects frozen or unresponsive nodes. Setting it auto-creates the safety monitor: if a node does not tick within the timeout, the watchdog triggers graduated degradation (warn → reduce rate → isolate → safe state).

Budget enforcement and deadline monitoring need no flag — they are always active for nodes that set .rate().

Sub-millisecond timeouts are rounded up to 1 ms with a warning (the config stores whole milliseconds). Duration::ZERO explicitly disables the watchdog.

Example:

let scheduler = Scheduler::new().watchdog(500_u64.ms());

Production Configuration

Chain builder methods for production deployments:

let mut scheduler = Scheduler::new()
    .tick_rate(1000_u64.hz())  // 1 kHz control loop
    .prefer_rt()               // RT priority + mlockall, degrade gracefully
    .blackbox(16)              // 16 MB flight recorder
    .watchdog(500_u64.ms());   // frozen-node watchdog

Node Management

add

pub fn add<N: Node + 'static>(&mut self, node: N) -> NodeBuilder<'_>

Adds a node using the fluent builder API. Returns a NodeBuilder for configuring the node.

Returns: NodeBuilder for chaining configuration

NodeBuilder Methods:

Execution Classes:

There is no .rt() method — the real-time class is derived automatically from the timing you declare. .rate() sets the period and auto-derives a budget (80% of the period) and deadline (95% of the period); calling .budget() or .deadline() without .rate() implicitly enables RT scheduling.

If no execution class is specified, the node stays on the scheduler's BestEffort set and is run by the ready-dispatch executor — in parallel across worker threads, ordered by the topic dependency graph. It is sequential only under .deterministic(true).

Example:

use horus::prelude::*;

scheduler.add(safety_monitor).order(0).budget(100.us()).build()?;
scheduler.add(motor_ctrl).order(5).rate(1000_u64.hz()).budget(500.us()).build()?;
scheduler.add(path_planner).order(10).compute().build()?;
scheduler.add(data_processor).order(20).on("sensor_data").build()?;
scheduler.add(cloud_uploader).order(100).async_io().rate(1_u64.hz()).build()?;
scheduler.add(sensor).order(50).rate(1000_u64.hz()).build()?;
scheduler.add(logger).order(200).build()?;

DurationExt — the trait behind .us(), .ms() and .hz() — and the Frequency type that .hz() returns both come from the prelude.

set_node_rate

pub fn set_node_rate(&mut self, name: &str, rate: Frequency) -> &mut Self

Sets per-node tick rate after registration. This is #[doc(hidden)] internal API — prefer .rate() on the node builder.

Example:

scheduler
    .set_node_rate("fast_sensor", 1000_u64.hz())  // 1 kHz
    .set_node_rate("slow_logger", 1_u64.hz());    // 1 Hz

Execution

run

pub fn run(&mut self) -> Result<()>

Runs all nodes until Ctrl+C or stop() is called. Blocking.

Returns: Result<()>

run_for

pub fn run_for(&mut self, duration: Duration) -> Result<()>

Runs all nodes for a specified duration.

Parameters:

• duration - How long to run

Returns: Result<()>

tick

pub fn tick(&mut self, node_names: &[&str]) -> Result<()>

Ticks specific nodes by name continuously.

tick_for

pub fn tick_for(&mut self, node_names: &[&str], duration: Duration) -> Result<()>

Ticks specific nodes for a specified duration.

stop

pub fn stop(&self)

Stops the scheduler.

is_running

pub fn is_running(&self) -> bool

Returns whether the scheduler is currently running.

current_tick

pub fn current_tick(&self) -> u64

Returns the current tick number.

Monitoring

OS Integration (Linux)

pin_to_cpu

pub fn pin_to_cpu(&self, cpu_id: usize) -> Result<()>

Pins scheduler thread to a specific CPU core. Linux only.

lock_memory

pub fn lock_memory(&self) -> Result<()>

Locks all memory pages to prevent page faults. Linux only. Requires CAP_IPC_LOCK capability or root.

Example

use horus::prelude::*;
use std::time::Duration;

fn main() -> Result<()> {
    // Configure for production: RT scheduling + flight recorder
    let mut scheduler = Scheduler::new()
        .prefer_rt()
        .blackbox(16);

    // Add nodes with execution classes using fluent API
    scheduler.add(EmergencyStopNode::new()?).order(0).budget(100.us()).build()?;
    scheduler.add(MotorController::new()?).order(5).rate(2000_u64.hz()).budget(500.us()).build()?;
    scheduler.add(PathPlanner::new()?).order(10).compute().build()?;
    scheduler.add(SensorReader::new()?).order(50).rate(1000_u64.hz()).build()?;
    scheduler.add(CloudSync::new()?).order(100).async_io().rate(1_u64.hz()).build()?;
    scheduler.add(DataLogger::new()?).order(200).rate(10_u64.hz()).build()?;

    // Run for 60 seconds
    scheduler.run_for(Duration::from_secs(60))?;

    Ok(())
}

hlog! Macro

The hlog! macro provides structured logging for nodes. Use it instead of println or other logging methods.

hlog!(level, "message", args...);

Log Levels

hlog!(debug, "Debug information: {}", value);
hlog!(info, "Node initialized");
hlog!(warn, "Potential issue: {}", issue);
hlog!(error, "Error occurred: {}", err);

Example

fn tick(&mut self) {
    self.publisher.send(data);
    hlog!(debug, "Published data: {:?}", data);
}

fn init(&mut self) -> Result<()> {
    hlog!(info, "Sensor initialized");
    Ok(())
}

Error

Unified error type for all HORUS operations. The prelude exports Result<T> and Error as short aliases, plus the HorusError name itself for matching on individual variants. HorusResult<T> still exists in horus_core::error but is not in the prelude — import it explicitly as use horus::HorusResult; if you need the long name.

Every payload is a structured sub-enum, not a bare String:

use horus::prelude::*;  // Imports Error, HorusError, Result

#[non_exhaustive]
#[derive(Debug, Error)]
pub enum HorusError {   // `Error` is an alias for this
    Io(std::io::Error),
    Config(ConfigError),
    Communication(CommunicationError),
    Node(NodeError),
    Memory(MemoryError),
    Serialization(SerializationError),
    NotFound(NotFoundError),
    Resource(ResourceError),        // AlreadyExists / PermissionDenied / Unsupported
    InvalidInput(ValidationError),
    Parse(ParseError),
    InvalidDescriptor(String),
    Transform(TransformError),
    Timeout(TimeoutError),
    Internal { message: String, file: &'static str, line: u32 },
    Contextual { message: String, source: Box<dyn std::error::Error + Send + Sync> },
}

Because the enum is #[non_exhaustive], a match over it needs a _ => catch-all arm.

Constructors and Helpers

impl HorusError {
    pub fn config<S: Into<String>>(msg: S) -> Self;                             // → Config(ConfigError::Other)
    pub fn node<S: Into<String>, T: Into<String>>(node: S, message: T) -> Self; // → Node(NodeError::Other)
    pub fn network_fault<S: Into<String>, T: Into<String>>(peer: S, reason: T) -> Self;
    pub fn help(&self) -> Option<&'static str>;
    pub fn severity(&self) -> Severity;
}

There are no communication, memory, invalid_input or internal constructors. Build those variants directly:

Error::InvalidInput(ValidationError::Other("speed must be positive".into()))
horus_internal!("Unexpected state: {:?}", state)

horus_internal! Macro

For internal errors with automatic file/line capture. It is not part of the prelude — import it explicitly:

use horus::horus_internal;
return Err(horus_internal!("Unexpected state: {:?}", state));
// Expands to: Error::Internal { message: "...", file: "src/foo.rs", line: 42 }

Result

Type alias for Results using Error. It comes from the prelude; as noted under Error, the long name HorusResult<T> does not, so import that one explicitly if you prefer it.

use horus::prelude::*;

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

Enums

NodeState

pub enum NodeState {
    Uninitialized,
    Initializing,
    Running,
    Stopping,
    Stopped,
    Error(String),
    Crashed(String),
}

HealthStatus

pub enum HealthStatus {
    Healthy = 0,   // operating normally
    Warning = 1,   // degraded performance (slow ticks, missed deadlines)
    Error = 2,     // errors occurring but still running
    Critical = 3,  // fatal errors, about to crash or unresponsive
    Unknown = 4,   // no heartbeat received (the default)
}

Health is driven by two independent inputs. The timing ladder moves a node down as it misses deadlines or trips its watchdog. Separately, a node whose tick() fails three times in a row stops being reported as healthy — any successful tick resets the run. A single failure can be a bad message or a one-off; a sustained run is a state change, and it is reflected in horus node info and horus node list.

Structs

NodeMetrics

Fields are private — read the values through the accessor methods:

pub struct NodeMetrics { /* fields private */ }

impl NodeMetrics {
    pub fn name(&self) -> &str;
    pub fn order(&self) -> u32;
    pub fn total_ticks(&self) -> u64;
    pub fn successful_ticks(&self) -> u64;
    pub fn failed_ticks(&self) -> u64;
    pub fn avg_tick_duration_ms(&self) -> f64;
    pub fn max_tick_duration_ms(&self) -> f64;
    pub fn min_tick_duration_ms(&self) -> f64;
    pub fn last_tick_duration_ms(&self) -> f64;
    pub fn messages_sent(&self) -> u64;      // always 0 — never written
    pub fn messages_received(&self) -> u64;  // always 0 — never written
    pub fn errors_count(&self) -> u64;
    pub fn warnings_count(&self) -> u64;
    pub fn uptime_seconds(&self) -> f64;
}

TopicMetrics

Fields are private — read the values through the accessor methods:

pub struct TopicMetrics { /* fields private */ }

impl TopicMetrics {
    pub fn messages_sent(&self) -> u64;      // verbose-logging path only
    pub fn messages_received(&self) -> u64;  // verbose-logging path only
    pub fn send_failures(&self) -> u64;
    pub fn recv_failures(&self) -> u64;
}

messages_sent and messages_received are incremented only on the #[cold] content- logging path, which runs while a topic's verbose flag is set from the horus monitor TUI. In a normal run both stay at 0 however much traffic the topic carries — the counter that is always maintained is messages_total in the shared-memory header, which is what horus topic list and horus topic hz report. send_failures (exposed as Topic::dropped_count()) and recv_failures are always live.

The two identically named methods on NodeMetrics above are different: nothing writes them at all, so they read 0 unconditionally.

TopicMetadata

pub struct TopicMetadata {
    pub topic_name: String,
    pub type_name: String,
}

Performance

Every topic is shared-memory backed, so every figure below is a cross-process number — there is no same-thread or intra-process fast path.

Topic automatically selects the backend based on message type and detected topology. These are the per-backend design figures for the topic backends themselves; for end-to-end measured latency see Benchmarks.

LogSummary Trait

Provides compact string representations for logging large data structures without cloning the full payload.

pub trait LogSummary {
    fn log_summary(&self) -> String;
}

Derive Macro

Use #[derive(LogSummary)] to auto-implement using Debug formatting:

#[derive(Debug, LogSummary)]
struct SensorReading {
    temperature: f32,
    humidity: f32,
}

Custom Implementation

For large types (images, point clouds), implement manually to show only metadata:

impl LogSummary for Image {
    fn log_summary(&self) -> String {
        format!("Image({}x{}, {:?})", self.width, self.height, self.encoding)
    }
}

Built-in implementations exist for GenericMessage, Tensor, HealthStatus, the primitive types (f32, f64, i32, i64, u32, u64, usize, bool, String), the image/pointcloud/depth descriptors, and all standard message types.

Rate

Drift-compensated rate limiter for controlling loop frequency in background threads.

let mut rate = Rate::new(100.0); // 100 Hz
loop {
    do_work();
    rate.sleep(); // Compensates for drift
}

Stopwatch

Simple elapsed time tracker.

let mut sw = Stopwatch::start();
expensive_operation();
println!("Took {:.2} ms", sw.elapsed_ms());
let lap = sw.lap(); // Returns elapsed and resets