Node macro reference

A node is a Rust struct annotated with #[cerulion_node(...)] paired with an adjacent impl block annotated with #[cerulion_node_impl]. Together they turn your struct into a node that cerulion node build compiles and cerulion graph run loads — you never call the generated code yourself. Import everything from the prelude:

use cerulion_core::prelude::*;

The prelude re-exports cerulion_node, cerulion_node_impl, NodeError, the event types, FillFrom/SliceSource, and the transport types.

The two macros

The macros are used as a pair:

#[cerulion_node(...)] on the struct declares the node type, its ports (via field attributes), and its trigger policy.
#[cerulion_node_impl] on an adjacent impl <Name> { … } block lets tick() read inputs and write outputs through ordinary field access (self.<port>.…) — reads and writes go directly to shared memory, zero-copy.

#[cerulion_node_impl] takes no arguments. The struct must appear before the impl block.

The impl block must define tick(&mut self) -> Result<(), NodeError>. It may also define init(&mut self, ctx: &mut NodeContext) -> Result<(), NodeError> (runs once before the first tick) and shutdown(&mut self) -> Result<(), NodeError> (runs once on a clean exit).

Complete example

use cerulion_core::prelude::*;
use native_ros2_messages::geometry_msgs::Vector3;
use native_ros2_messages::sensor_msgs::LaserScan;

#[cerulion_node(period_ms = 10)]      // period-driven; reads the latest scan each tick
#[derive(Default)]
struct SafetyController {
    #[input(depth = 1)]               // latest-value context input
    scan: LaserScan,

    #[output]                         // Vector3 is a fixed-only schema (x/y/z: f64)
    linear_velocity: Vector3,
}

#[cerulion_node_impl]
impl SafetyController {
    fn tick(&mut self) -> Result<(), NodeError> {
        // Variable-length field read through its typed accessor:
        let obstacle_close = self.scan.ranges().iter().any(|&r| r < 0.5);
        // Field writes go straight to shared memory:
        self.linear_velocity.x = if obstacle_close { 0.0 } else { 0.3 };
        Ok(())
    }
}

Build it with cerulion node build safety_controller, then wire it into a graph with cerulion node stage safety_controller.

Node-level attributes

Exactly one trigger policy applies — either an attribute below, or an inferred data trigger from a field marked #[input(trigger)]. A node that declares ports but neither is a compile error.

Attribute	Semantics
`period_ms = N`	Fire every N ms. N must be > 0.
`sync_window_ms = N`	Bounded sync: fire once per complete set of `#[input(trigger)]` messages that arrived within an N-ms window. Requires ≥2 trigger inputs, all wired in the graph.
`unbounded_sync`	Sync with no time bound. Requires ≥2 trigger inputs. Suited to batch and offline fusion; control loops want `sync_window_ms` for a bounded latency.
`external`	Self-triggering ingress node. Requires an `external_source()` method returning `ExternalSource::{Fd, Blocking, HostDriven}`. Rejected under `--time-source virtual`.
`tick_within_ms = N`	QoS, not a trigger. Per-node tick-duration deadline; a slower tick increments `tick_within_missed_count` and warns with `node_id`, `tick_within_ms`, and `elapsed_ms`.
`throttle_ms = N`	Rate cap, not a trigger. Defers the node’s tick while `now − last_fire < N ms`.

Mutual exclusion

unbounded_sync excludes sync_window_ms, period_ms, and external. throttle_ms excludes period_ms (a period already pins the rate). tick_within_ms stacks with every trigger policy.

Rejected attributes

Attribute	Reason
`type_name`	Not an attribute — the node type is the folder name.
`inputs(...)` / `outputs(...)`	Ports are declared as field attributes, not macro arguments.
`deadline_ms = N`	Use `#[input(expect_within_ms = N)]` or `#[output(promise_within_ms = N)]`.

`#[input(...)]` attribute

Declares an input port; the field type is the schema.

Key	Default	Meaning
`trigger`		This input fires the node (data trigger, or a member of a sync set).
`depth = N`	`10`	Receive queue depth in messages. Maximum 64 — a consumer that needs more in-flight messages than that is permanently slower than its producer, so reach for a backpressure policy instead.
`backpressure = drop_oldest \| sample(N) \| block`	`drop_oldest`	What happens when the queue would overflow. `block` needs every consumer of the topic to declare it and an in-graph producer; a hand-written `process_groups:` may split a `block` edge across processes when it is creditable. See Backpressure and deadlines.
`expect_within_ms = N`		QoS: if N ms elapse with no fresh data on this input, increment `expect_within_missed_count` and warn. Independent of `trigger`.

That is the complete surface: trigger, depth, backpressure, expect_within_ms. Any other key fails the build. The diagnostic names the unrecognized key and lists the accepted set.

Trigger inputs versus context inputs

A plain #[input] that is not the node’s trigger is a latest-value context read: in tick() it yields the most recent message on that topic, held across ticks. Until it has delivered at least once, the tick is a no-op — Cerulion never fabricates a default value for data that has not arrived. That means a 1 Hz map does not starve a 100 Hz consumer, and it also means a producer that stalls without disconnecting keeps serving its last value; add expect_within_ms if you need to notice.

`#[output(...)]` attribute

Form	Meaning
`#[output]`	The canonical form, for every schema — fixed-only or variable. There is nothing to declare per field.
`#[output(promise_within_ms = N)]`	QoS: if N ms elapse with no publish on this output, increment `promise_within_missed_count` and warn.

#[output] takes no field list. A form such as #[output(data, encoding)] is rejected with guidance to assign self.<port>.<field> = expr instead. Only the two forms above are accepted.

Inside `tick()`

Pattern	Effect
`self.<port>.<fixed_field>`	Reads a fixed field directly from shared memory. Zero copy.
`self.<port>.<var_field>()`	Reads a variable-length field through its accessor method: `self.scan.ranges()` returns `&[f32]`; string fields return a `Result<&str, _>`.
`self.<port>.<field> = expr`	One rule for fixed and variable output fields alike: the write lands in shared memory, and the macro makes it fallible for you.
`self.<port>.<nested>.<leaf> = expr`	Nested writes work to any depth, for example `self.detection.pose.position.x = 0.5;`.
`self.<port>.<var_field>.fill_from(producer)?`	Zero-copy producer write. See below.
`self.<port>.emit()?`	Publishes a fieldless message (`std_msgs/Empty`). Exists only on zero-field schemas.
`self.<other_field>`	Plain Rust state. Keep per-tick state here.
`self.now_ns()`	Time in ns from the run’s active clock — the read to use in node logic, and the only one that is replay-safe.
`self.real_ns()`	Raw monotonic hardware ns. Non-deterministic; for benchmarks only.
`self.virt_ns()` / `self.ext_ns()`	`Some(ns)` only under the virtual / external clock; `None` otherwise.
`self.request_shutdown()`	Asks the runtime to drain and exit cleanly.

Publishing rules worth memorizing

Only the outputs you write are published. An output your tick never touches is not published and costs nothing.
A tick that returns Err publishes nothing. Every output it had written is released without sending.
A variable-schema output must have all of its variable fields written on a tick that publishes it, at every nesting depth, or the frame is discarded with a loud error.
Do not read your own output fields — reading one loans a frame you did not mean to send. Keep state in plain struct fields.

Helpers and closures must be fallible

Every port-field assignment carries an injected ?, so any method or closure that writes a port field has to return a Result:

// Rejected: a `()` helper writing a port field.
fn set_speed(&mut self) { self.cmd_vel.x = 0.3; }

// Correct: return Result and propagate at the call site.
fn set_speed(&mut self) -> Result<(), NodeError> {
    self.cmd_vel.x = 0.3;
    Ok(())
}

The same applies inside closures: hoist the value out, or use a fallible combinator such as try_for_each.

Port fields cannot be used inside a macro argument (format!, tracing::info!). Assign to a local first, then log the local.

`fill_from`: zero-copy producer writes

For producers that write into a destination buffer — camera drivers, codecs, file readers, sockets — fill_from passes the shared-memory destination (&mut [T]) straight to the producer.

// Closure form: the producer writes directly into shared memory.
self.image.data.fill_from(|buf: &mut [u8]| {
    let n = self.driver.read_frame_into(buf).map_err(|e| TransportError::NodeError {
        node_id: "camera".into(),
        reason: e.to_string(),
    })?;
    Ok(n)  // bytes written; only the first n are published
})?;

// Slice form: copy from an existing &[u8] without an intermediate Vec.
self.image.data.fill_from(SliceSource::new(&raw_bytes))?;

Accepted producers are a closure FnMut(&mut [T]) -> Result<usize, TransportError>, SliceSource::new(slice), or any type implementing FillFrom<T>. Typed arrays get the element type directly, for example &mut [f32] for a float32[] field. If the producer returns Err, nothing is published that tick.

`#[on_event]` callbacks

Annotate a method in the #[cerulion_node_impl] block. The event parameter type selects the event, and the filter must name a declared port of the matching scope. Handlers run at the end of a successful tick, in declaration order.

Filter	Event types
`#[on_event(input = "<port>")]`	`BackpressureEvent`, `ExpectWithinEvent`, `LivelinessEvent`
`#[on_event(output = "<port>")]`	`PromiseWithinEvent`

#[cerulion_node_impl]
impl FusionNode {
    fn tick(&mut self) -> Result<(), NodeError> { Ok(()) }

    #[on_event(input = "imu")]
    fn on_imu_overdue(&mut self, event: ExpectWithinEvent) {
        tracing::warn!(input = %event.input_name, elapsed_ms = event.elapsed_ms, "IMU data overdue");
    }
}

All four events are edge-triggered: one event per regime, not one per occurrence. Two handlers on the same port are fine if their event types differ; two with the same port and type are a compile error, and a mismatched scope is a compile error too.

Handlers dispatch on the node’s own tick. A purely data-triggered node stops ticking when its input goes silent, so it may never run its LivelinessEvent Lost handler. Give such a node a period trigger if you depend on hearing about disconnects.

Error type

tick(), init(), and shutdown() return Result<(), NodeError>. The variants you construct are NodeError::Logic(String), NodeError::InvalidInput { input, reason }, NodeError::Fatal(String), and NodeError::Custom(Box<dyn Error + Send + Sync>). Transport and I/O errors convert automatically, so ? works on them.

Trigger policies

The --policy grammar and the defaulting matrix.

Backpressure and deadlines

Pick a queue policy and set arrival deadlines.

Define a node

The end-to-end workflow for a new node type.

Get started

Concepts

Guides

Studio and Hindsight

Reference

Community

The two macros

Complete example

Node-level attributes

Mutual exclusion

Rejected attributes

`#[input(...)]` attribute

Trigger inputs versus context inputs

`#[output(...)]` attribute

Inside `tick()`

Publishing rules worth memorizing

Helpers and closures must be fallible

`fill_from`: zero-copy producer writes

`#[on_event]` callbacks

Error type

Trigger policies

Backpressure and deadlines

Define a node

​The two macros

​Complete example

​Node-level attributes

​Mutual exclusion

​Rejected attributes

​#[input(...)] attribute

​Trigger inputs versus context inputs

​#[output(...)] attribute

​Inside tick()

​Publishing rules worth memorizing

​Helpers and closures must be fallible

​fill_from: zero-copy producer writes

​#[on_event] callbacks

​Error type

​Related

Trigger policies

Backpressure and deadlines

Define a node

The two macros

Complete example

Node-level attributes

Mutual exclusion

Rejected attributes

`#[input(...)]` attribute

Trigger inputs versus context inputs

`#[output(...)]` attribute

Inside `tick()`

Publishing rules worth memorizing

Helpers and closures must be fallible

`fill_from`: zero-copy producer writes

`#[on_event]` callbacks

Error type

Related