mt_pubsub

A simple and deterministic Pub/Sub implementation with Minot primitives.

Part of the minot family of crates.

[dependencies]
mt_pubsub = "0.9.0"
tokio = { version = "1.49", features = ["full"] }

Since you probably want to use existing ROS2 message definitions, you can also add the following crate which is auto-generated from the Jazzy release.

[dependencies]
ros2-interfaces-jazzy-rkyv = { version = "0.0.5", features = [
  "std_msgs", # add more here
] }

Learn more on how to use it in your Code by visiting the Minot docs.

Running the Examples

First, start the coordinator:

cargo run --bin minot -- coord

Then run the publisher and subscriber in separate terminals:

# Terminal 2
cargo run --example pub

# Terminal 3
cargo run --example sub

If you have multiple Minot networks on the same physical network, use the MINOT_DOMAIN_ID environment variable to prevent them from connecting to each other.

Delivery modes

Qos sets a node's delivery mode. It encodes two independent things, and it is worth keeping them apart: how Zenoh carries the bytes, and what the rest of the system does when the node stops answering.

Reliable is the default and the strict one: peers heartbeat-monitor each other, and a node that stops answering tears the whole run down. That is what you want for a deterministic replay where a missing participant invalidates the result.

TryReliable keeps reliable delivery and ordering but drops the fatal part. It behaves like a ROS 2 DDS RELIABLE subscriber: a write that cannot land within its deadline fails on its own instead of taking the participant down, and no peer is torpedoed. Sends are dispatched off the caller's thread, so a slow or roaming link cannot wedge a publisher's loop.

BestEffort tries once and moves on.

Note that reliable wire delivery is close to free over TCP, which already retransmits and orders. The meaningful difference between Reliable and TryReliable is the failure policy, not the wire.

Set the mode on the node:

use mt_pubsub::{Node, NodeConfig, Qos};

let node = Node::create(NodeConfig::new("viewer").mode(Qos::TryReliable)).await?;

Or per topic, which overrides the node's mode for that subscription only:

let points = node
    .create_subscriber::<PointCloud2>("/ouster/points".to_owned(), 10, Qos::TryReliable)
    .await?;

A best-effort publisher drops samples, so a subscriber that expects delivery (Reliable or TryReliable) is rejected at registration. Pair a best-effort publisher with best-effort subscribers.

The TryReliable budget is tunable in mt_sea: TRY_RELIABLE_SEND_BUDGET_MS (default 3000) bounds one delivery, TRY_RELIABLE_ATTEMPT_TIMEOUT_MS (500) bounds a single attempt inside it, and TRY_RELIABLE_RETRY_BACKOFF_MS (50) spaces the retries.

Shared memory

The shm feature uses Zenoh shared memory for same-device messages of at least 1 MiB. SHM capability is negotiated per Zenoh connection. If either local process has SHM disabled or unavailable, SHM-backed messages are carried as ordinary network bytes on that connection. Enabling SHM therefore remains compatible with peers built or started without SHM.

Defaults can be changed with:

• MINOT_SHM_SIZE — initial pool size in bytes (default: 16 MiB)
• MINOT_SHM_THRESHOLD — minimum message size attempted through SHM (default: 1 MiB)
• MINOT_SHM_MAX_MESSAGE_SIZE — largest message attempted through SHM (default: 64 MiB)
• MINOT_SHM_ALLOCATION_TIMEOUT_MS — allocation deadline before fallback (default: 250 ms)
• MINOT_SHM_STATS_INTERVAL_SECS — interval for non-empty SHM/non-SHM traffic summaries from the minot binary (default: 30 seconds)
• MINOT_SHM_DISABLED=1 — always use regular Zenoh transport

The minot binary exposes the same controls as --shm-size, --shm-threshold, --shm-max-message-size, --shm-allocation-timeout-ms, and --no-shm.

Run the reliability suite and manual benchmark with:

cargo test -p mt_pubsub --test shm_reliability --features shm
cargo test -p mt_pubsub --test shm_reliability --features shm --release \
  shm_large_packet_benchmark -- --exact --ignored --nocapture

Borrowed archived messages

Subscriber::next() returns an owned, deserialized message. Code that can work with rkyv's archived representation can avoid that conversion with next_archived():

if let Some(message) = subscriber.next_archived().await {
    let archived = message.archived();
    // Read archived fields while `message` keeps the backing buffer alive.
}

Call message.deserialize() when an existing API still requires an owned value.