[Bug]: Pub/Sub crashes across schedulers when using Image messages (separate processes)

[gelanchez]

Bug Description

When using HORUS pub/sub across two different files, each running its own scheduler, the subscriber node crashes when publishing/subscribing Image messages.

The exact same nodes work correctly when both publisher and subscriber are run in the same binary with a single scheduler.

Steps to Reproduce

1. Create a publisher node that publishes an Image message on a topic.
2. Create a subscriber node that subscribes to the same topic.
3. Compile and run each node as a separate binary, each with its own Scheduler.
4. Observe that the subscriber crashes at runtime.
5. Run both nodes in the same binary and scheduler → the image is received correctly.

Expected Behavior

Image pub/sub works across schedulers/processes. If an error happens, a runtime error should be raised.

Actual Behavior

Process [node] exited with code: -1

HORUS Version

0.1.7

Operating System

Ubuntu 24.04

Rust Version

rustc 1.92.0

Language

Rust

Minimal Code Example

test.rs

use horus::prelude::*;

node! {
    TestSendNode {
        name: "test_send_node",
        rate 1.0
        pub {
            test_pub: Image -> "test_send.image"
        }
        tick(ctx) {
            let width = 324;
            let height = 244;
            let pixels = vec![0u8; width * height * 3]; // Empty image
            let image = Image::new(width as u32, height as u32, ImageEncoding::Rgb8, pixels);

            let _ = self.test_pub.send(image, &mut ctx);
        }
    }
}

fn main() -> Result<()> {
    let mut scheduler = Scheduler::new().name("test_scheduler");

    scheduler.add(Box::new(TestSendNode::new()), 1, Some(true));

    // Run the scheduler loop - continuously ticks all nodes
    let _ = scheduler.run();
    Ok(())
}

test_receive.rs

use horus::prelude::*;

node! {
    TestReceiveNode {
        name: "test_receive_node",
        rate 5.0
        sub {
            test_sub: Image -> "test_send.image"
        }
        tick(ctx) {
            if let Some(image) = self.test_sub.recv(&mut ctx) {
                if image.is_valid() {
                    println!("Received image");
                }
            }
        }
    }
}

fn main() -> Result<()> {
    let mut scheduler = Scheduler::new().name("test_receive_scheduler");

    scheduler.add(Box::new(TestReceiveNode::new()), 1, Some(true));

    // Run the scheduler loop - continuously ticks all nodes
    let _ = scheduler.run();
    Ok(())
}

Relevant Logs

Process [test_receive] exited with code: -1

Additional Context

No response

[gelanchez]

A workaround on this issue is having the nodes as modules and using them inside a main, and therefore with the same scheduler. A disadvantage would be when sharing data with other pieces of the code that must be run from the main thread (e.g., egui), and therefore require a separate file / main.

[whochangedmycode]

Root Cause Analysis

After investigating the codebase, I found the issue:

Problem

The AdaptiveTopic uses a default slot size of 8KB (DEFAULT_SLOT_SIZE = 8 * 1024) for serialized (non-POD) messages. Your Image message is:

• 324 × 244 × 3 = 237,168 bytes (~232KB)

This is 29x larger than the slot size!

What Happens

1. Publisher tries to send Image via bincode serialization
2. Send path checks: if serialized.len() > max_data_size { return Err(msg); }
3. Send fails silently (you use .ok() which discards the error)
4. Subscriber reads uninitialized/stale shared memory data
5. Crash occurs during deserialization or memory access

The crash occurs because:

When the sender fails, the shared memory slot contains garbage data. The receiver:

1. Reads a garbage sequence number
2. If it accidentally matches tail + 1, proceeds to read
3. Reads garbage length value (could be huge)
4. Creates an invalid slice and crashes

Workaround (Immediate)

Use .unwrap() or .expect() on send to detect failures:

self.test_pub.send(image, &mut ctx).expect("Image too large for topic");

Fix Needed

The Topic API needs a way to configure slot size for large messages:

// Proposed API
let topic: Topic<Image> = Topic::new("camera")
    .with_slot_size(256 * 1024)  // 256KB for images
    .build()?;

Or auto-detect based on type size (though this is tricky for variable-length types like Image).

Additional Issue

The code should also add better bounds checking in recv to prevent crashes when reading corrupted data.

[neos-builder]

I will add the API for slot size and to documentation, future work might be making internal topic have smart detection for large message in the library

[gelanchez]

Thanks for the investigation. I agree that's one of the issues and needs to be addressed in order to use big images.

However, even with small images, the node crashes when running from different schedulers / horus instances, maybe due to the lack of synchronization. See this code (I've updated it to make it compile with the latest version of HORUS):

test_send.rs

use horus::prelude::*;

node! {
    TestSendNode {
        name: "test_send_node",
        rate 1.0
        pub {
            test_pub: Image -> "test_send.image",
        }
        init(ctx) {
            self.test_pub = Topic::<Image>::with_capacity("test_send.image", 1 << 16)?;
            Ok(())
        }
        tick(ctx) {
            let width = 3;//324;
            let height = 2;//244;
            let pixels = vec![0u8; width * height * 3]; // Empty image
            let image = Image::new(width as u32, height as u32, ImageEncoding::Rgb8, pixels);

            match self.test_pub.send(image) {
                Ok(()) => println!("Sent image"),
                Err(_e) => println!("Error"),
            }
        }
    }
}

fn main() -> Result<()> {
    let mut scheduler = Scheduler::new().name("test_scheduler");
    scheduler.add(TestSendNode::new()).order(0).done();
    let _ = scheduler.run();
    Ok(())
}

test_receive.rs

use horus::prelude::*;

node! {
    TestReceiveNode {
        name: "test_receive_node",
        rate 1.0
        sub {
            test_sub: Image -> "test_send.image"
        }
        tick(ctx) {
            if let Some(image) = self.test_sub.recv() {
                if image.is_valid() {
                    println!("Received image");
                }
            }
            else {
                println!("No image");
            }
        }
    }
}

fn main() -> Result<()> {
    let mut scheduler = Scheduler::new().name("test_receive_scheduler");
    scheduler.add(TestReceiveNode::new()).order(0).done();
    let _ = scheduler.run();
    Ok(())
}

Run it with horus run nodes/test_send.rs nodes/test_receive.rs