diff --git a/crates/hiroz-tests/Cargo.toml b/crates/hiroz-tests/Cargo.toml index 2eb259978..2e59c01ef 100644 --- a/crates/hiroz-tests/Cargo.toml +++ b/crates/hiroz-tests/Cargo.toml @@ -32,6 +32,7 @@ serde = { workspace = true } # For CDR serialization in dds_interop tests serde_json = "1.0" # For parsing --json output in the hu plugin tests [features] +# Forwards to hiroz; see that crate. Off by default. default = [] ros-msgs = [ "dep:hiroz-msgs", diff --git a/crates/hiroz-tests/tests/intra_process.rs b/crates/hiroz-tests/tests/intra_process.rs new file mode 100644 index 000000000..54a852998 --- /dev/null +++ b/crates/hiroz-tests/tests/intra_process.rs @@ -0,0 +1,1468 @@ +//! Intra-process `Arc` delivery — does it really skip the serialization? +//! +//! # The assertion that matters is `Arc::ptr_eq` +//! +//! "The message arrived and its contents match" is satisfied by a CDR round +//! trip just as well as by a pointer move, so it proves nothing about which +//! path ran. Pointer identity between the `Arc` that was published and the +//! `Arc` that was received cannot survive a serialization boundary: encoding +//! and decoding necessarily produces a new allocation. That is the property +//! under test. +//! +//! # What each test pins +//! +//! | test | property | +//! |---|---| +//! | `same_arc_reaches_a_same_session_subscriber` | pointer identity — no encode, no copy | +//! | `intra_process_only_publisher_does_not_reach_another_context` | nothing goes on the wire, with a control proving the wire works | +//! | `a_different_rust_type_on_the_same_topic_is_not_delivered` | the `TypeId` gate holds | +//! | `dropping_the_subscriber_unregisters_it` | no delivery into a dead subscriber | +//! | `a_pooled_payload_buffer_is_written_in_place_and_reused` | one buffer serves many sends | +//! | `a_plain_publish_reaches_a_shared_callback_subscriber` | #39 — no silent same-session loss | +//! | `publish_shared_without_a_locality_restriction_arrives_once` | #39 — and no duplicate either | +//! | `a_self_publishing_callback_does_not_recurse_without_bound` | #40 — a cycle is refused, not fatal | +//! | `a_plain_publisher_takes_the_wire_for_an_ordinary_subscriber` | #36 — no drop when nobody is on the bus | +//! | `the_wire_is_used_when_a_subscriber_is_off_session` | #36 — and the bus is not, so no duplicate | +//! | `a_sole_receiver_is_given_the_message_to_own` | #36 — the move, not a shared Arc | + +mod common; + +use std::{ + sync::{ + Arc, Mutex, OnceLock, + atomic::{AtomicUsize, Ordering}, + }, + thread, + time::{Duration, Instant}, +}; + +use common::*; +use hiroz::{ + Builder, + local_bus::{Delivery, Published}, +}; +use hiroz_msgs::std_msgs::{Int32, String as RosString}; + +/// The erased republish hook the depth tests install. +/// +/// Named so the signature does not have to be spelled inline: `ZPub` carries a +/// serializer parameter, and writing it out at the binding is what tripped +/// `clippy::type_complexity`. +type EchoPublisher = Box hiroz::Result + Send + Sync>; + +const DELIVERY_DEADLINE: Duration = Duration::from_secs(5); + +/// Ceiling on the self-publishing echo, so neither direction of the +/// recursion test can run forever. Far above the delivery-depth bound. +const ECHO_CAP: usize = 64; + +fn wait_until(f: impl Fn() -> bool) -> bool { + let start = Instant::now(); + while start.elapsed() < DELIVERY_DEADLINE { + if f() { + return true; + } + thread::sleep(Duration::from_millis(20)); + } + f() +} + +#[test] +fn same_arc_reaches_a_same_session_subscriber() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node_tx = ctx.create_node("zc_tx").build()?; + let node_rx = ctx.create_node("zc_rx").build()?; + + let received: Arc>>> = Arc::new(Mutex::new(Vec::new())); + let sink = received.clone(); + let _sub = node_rx + .create_sub::("shared") + .build_with_shared_callback(move |msg: Arc| { + sink.lock().expect("poisoned").push(msg); + })?; + + let publisher = node_tx + .create_pub::("shared") + .with_intra_process_only() + .build()?; + + let sent = Arc::new(RosString { + data: "no copies please".to_owned(), + }); + let delivered = publisher.publish_shared(sent.clone())?; + assert_eq!( + delivered, + Published::Bus(Delivery::Sent(1)), + "expected exactly one local subscriber" + ); + + // Delivery is synchronous on this thread, so no waiting is needed — but + // assert on the count first so a failure says "nothing arrived" rather than + // panicking on an empty vec. + let got = received.lock().expect("poisoned"); + assert_eq!(got.len(), 1, "subscriber did not receive the message"); + assert!( + Arc::ptr_eq(&sent, &got[0]), + "received a different allocation — the message went through a copy or a \ + serialization round trip, which is exactly what this path exists to avoid" + ); + Ok(()) +} + +#[test] +fn intra_process_only_publisher_does_not_reach_another_context() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx_tx = create_hiroz_context_with_router(&router)?; + let ctx_other = create_hiroz_context_with_router(&router)?; + + let node_tx = ctx_tx.create_node("zc_tx").build()?; + let node_other = ctx_other.create_node("zc_other").build()?; + + let local_hits = Arc::new(AtomicUsize::new(0)); + let control_hits = Arc::new(AtomicUsize::new(0)); + let (c1, c2) = (local_hits.clone(), control_hits.clone()); + + let _s1 = node_other + .create_sub::("zc_local") + .build_with_callback(move |_m: RosString| { + c1.fetch_add(1, Ordering::SeqCst); + })?; + let _s2 = node_other + .create_sub::("zc_control") + .build_with_callback(move |_m: RosString| { + c2.fetch_add(1, Ordering::SeqCst); + })?; + + let pub_local = node_tx + .create_pub::("zc_local") + .with_intra_process_only() + .build()?; + let pub_control = node_tx.create_pub::("zc_control").build()?; + + wait_for_ready(Duration::from_millis(500)); + + let msg = Arc::new(RosString { + data: "hello".to_owned(), + }); + for _ in 0..5 { + pub_local.publish_shared(msg.clone())?; + pub_control.publish(&msg)?; + } + + // The control must cross. Without it, a zero on `zc_local` would be + // indistinguishable from a broken router or a wrong topic. + assert!( + wait_until(|| control_hits.load(Ordering::SeqCst) >= 5), + "control publisher did not reach the other context ({} of 5) — this run \ + proves nothing about the intra-process path", + control_hits.load(Ordering::SeqCst) + ); + assert_eq!( + local_hits.load(Ordering::SeqCst), + 0, + "an intra-process-only publisher put something on the wire" + ); + Ok(()) +} + +#[test] +fn a_different_rust_type_on_the_same_topic_is_not_delivered() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("zc_types").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node + .create_sub::("typed") + .build_with_shared_callback(move |_m: Arc| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + // Positive control: without it, this test passes against a bus that + // delivers nothing at all, which is the failure mode it exists to detect. + let ok_hits = Arc::new(AtomicUsize::new(0)); + let ok = ok_hits.clone(); + let _sub_ok = node + .create_sub::("typed_control") + .build_with_shared_callback(move |_m: Arc| { + ok.fetch_add(1, Ordering::SeqCst); + })?; + let control_pub = node + .create_pub::("typed_control") + .with_intra_process_only() + .build()?; + assert_eq!( + control_pub.publish_shared(Arc::new(Int32 { data: 7 }))?, + Published::Bus(Delivery::Sent(1)), + "the control did not deliver: the bus is dead, so the assertion below proves nothing" + ); + assert_eq!( + ok_hits.load(Ordering::SeqCst), + 1, + "control subscriber not called" + ); + + // Same topic, different concrete Rust type. + let publisher = node + .create_pub::("typed") + .with_intra_process_only() + .build()?; + let delivered = publisher.publish_shared(Arc::new(RosString { + data: "wrong type".to_owned(), + }))?; + + assert_eq!( + delivered, + Published::Bus(Delivery::NoTaker), + "delivered across a type mismatch" + ); + assert_eq!( + hits.load(Ordering::SeqCst), + 0, + "an Int32 subscriber was handed a String — the TypeId gate is not holding" + ); + Ok(()) +} + +#[test] +fn dropping_the_subscriber_unregisters_it() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("zc_drop").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let sub = node + .create_sub::("dropme") + .build_with_shared_callback(move |_m: Arc| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("dropme") + .with_intra_process_only() + .build()?; + let msg = Arc::new(RosString { + data: "x".to_owned(), + }); + + // Positive control: it is registered right now. Without this the assertion + // below would pass just as well against a subscriber that never worked. + assert_eq!( + publisher.publish_shared(msg.clone())?, + Published::Bus(Delivery::Sent(1)) + ); + assert_eq!(hits.load(Ordering::SeqCst), 1); + + drop(sub); + + assert_eq!( + publisher.publish_shared(msg)?, + Published::Bus(Delivery::NoTaker), + "the bus still holds a registration for a dropped subscriber" + ); + assert_eq!( + hits.load(Ordering::SeqCst), + 1, + "a dropped subscriber was called" + ); + Ok(()) +} + +/// #39 — an ordinary publisher must reach a shared-callback subscriber. +/// +/// The bus subscriber's wire half used to be forced to `Locality::Remote`, +/// which silently discarded every same-session `publish`. Nothing errored and +/// the graph showed both endpoints matched, so only an assertion on delivery +/// can catch it. +#[test] +fn a_plain_publish_reaches_a_shared_callback_subscriber() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node_tx = ctx.create_node("plain_tx").build()?; + let node_rx = ctx.create_node("plain_rx").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node_rx + .create_sub::("plain_to_shared") + .build_with_shared_callback(move |_m: Arc| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + // An ordinary publisher: no locality, no bus involvement, nothing special. + let publisher = node_tx.create_pub::("plain_to_shared").build()?; + wait_for_ready(Duration::from_millis(500)); + + publisher.publish(&RosString { + data: "over the wire".to_owned(), + })?; + + assert!( + wait_until(|| hits.load(Ordering::SeqCst) >= 1), + "a same-session plain publish never reached the shared-callback subscriber" + ); + assert_eq!( + hits.load(Ordering::SeqCst), + 1, + "delivered more than once — the wire and the bus both fired" + ); + Ok(()) +} + +/// #39 — and the duplicate the old filter existed to prevent must stay prevented. +/// +/// A `publish_shared` on a publisher with no locality restriction reaches the +/// subscriber over the wire. It must not *also* arrive over the bus. +#[test] +fn publish_shared_without_a_locality_restriction_arrives_once() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node_tx = ctx.create_node("once_tx").build()?; + let node_rx = ctx.create_node("once_rx").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let seen: Arc>>> = Arc::new(Mutex::new(Vec::new())); + let h = hits.clone(); + let sink = seen.clone(); + let _sub = node_rx + .create_sub::("exactly_once") + .build_with_shared_callback(move |m: Arc| { + h.fetch_add(1, Ordering::SeqCst); + sink.lock().expect("poisoned").push(m); + })?; + + // A plain publisher: no locality, no flag, nothing telling it what to do. + let publisher = node_tx.create_pub::("exactly_once").build()?; + wait_for_ready(Duration::from_millis(500)); + + let sent = Arc::new(RosString { + data: "once please".to_owned(), + }); + let delivered = publisher.publish_shared(sent.clone())?; + // The wire, because this publisher asserted nothing about its audience. + // Inferring "everyone is local" from the ROS graph is unsound: it cannot + // see a plain zenoh subscriber. See #134. + assert_eq!( + delivered, + Published::Wire, + "the bus was taken without the caller asserting the audience" + ); + + assert!( + wait_until(|| hits.load(Ordering::SeqCst) >= 1), + "the message did not arrive at all" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!( + hits.load(Ordering::SeqCst), + 1, + "delivered twice — once over the bus and once over the wire" + ); + // Deliberately NOT asserting pointer identity: it arrived over the wire, so + // it is a decoded copy. The property here is exactly-once, not zero-copy. + let got = seen.lock().expect("poisoned"); + assert_eq!(got.len(), 1, "subscriber did not receive the message"); + Ok(()) +} + +/// #40 — a callback that publishes onto its own topic must not recurse forever. +/// +/// Delivery is inline on the publishing thread, so this is direct recursion. +/// On the wire the same shape is an endless stream of messages, which is +/// observable and survivable; here it used to end in a stack overflow. The bus +/// now refuses past a fixed depth, so the chain terminates and the test +/// returns rather than dying with SIGSEGV. +#[test] +fn a_self_publishing_callback_does_not_recurse_without_bound() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("echo_node").build()?; + + let publisher = Arc::new( + node.create_pub::("echo") + .with_intra_process_only() + .build()?, + ); + let echo = publisher.clone(); + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node + .create_sub::("echo") + .build_with_shared_callback(move |m: Arc| { + let seen = h.fetch_add(1, Ordering::SeqCst); + // ECHO_CAP so that reverting the depth guard fails this test by + // assertion rather than by stack overflow, which aborts the whole + // binary and names no property at all. + if seen < ECHO_CAP { + // Republish onto the very topic this callback serves. + let _ = echo.publish_shared(m); + } + })?; + + publisher.publish_shared(Arc::new(RosString { + data: "round and round".to_owned(), + }))?; + + let seen = hits.load(Ordering::SeqCst); + assert!(seen >= 1, "the callback never ran"); + assert_eq!( + seen, + hiroz::local_bus::MAX_DELIVERY_DEPTH as usize, + "expected exactly the depth bound; a change to MAX_DELIVERY_DEPTH must \ + update this test rather than slip past a loose ceiling" + ); + assert!( + seen <= 16, + "delivery recursed {seen} deep — the depth guard did not hold" + ); + Ok(()) +} + +/// #36 — a publisher without the flag must read its audience off the graph. +/// +/// The prototype was *told* whether to use the wire, so `publish_shared` on a +/// publisher whose subscriber was an ordinary one delivered nothing at all. +/// Now the publisher asks: no bus subscriber means the wire carries it. +#[test] +fn a_plain_publisher_takes_the_wire_for_an_ordinary_subscriber() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node_tx = ctx.create_node("ask_tx").build()?; + let node_rx = ctx.create_node("ask_rx").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + // An ORDINARY subscriber: not on the bus, expects decoded bytes. + let _sub = node_rx + .create_sub::("ask_graph") + .build_with_callback(move |_m: RosString| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + // No locality, no flag. The publisher has to work out the audience. + let publisher = node_tx.create_pub::("ask_graph").build()?; + wait_for_ready(Duration::from_millis(500)); + + let delivered = publisher.publish_shared(Arc::new(RosString { + data: "who is listening".to_owned(), + }))?; + assert_eq!( + delivered, + Published::Wire, + "the bus took a message its subscriber cannot decode" + ); + + assert!( + wait_until(|| hits.load(Ordering::SeqCst) >= 1), + "publish_shared reached nobody: the publisher neither used the bus nor the wire" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!(hits.load(Ordering::SeqCst), 1, "delivered more than once"); + Ok(()) +} + +/// #36 — an off-session subscriber forces the wire, and then the bus stays out. +/// +/// Both paths at once would deliver twice to a same-session bus subscriber. +/// The publisher takes the wire alone, so the local subscriber gets exactly one +/// copy — decoded rather than shared, which is the price of a remote audience. +#[test] +fn the_wire_is_used_when_a_subscriber_is_off_session() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx_tx = create_hiroz_context_with_router(&router)?; + let ctx_far = create_hiroz_context_with_router(&router)?; + + let node_tx = ctx_tx.create_node("mix_tx").build()?; + let node_near = ctx_tx.create_node("mix_near").build()?; + let node_far = ctx_far.create_node("mix_far").build()?; + + let near = Arc::new(AtomicUsize::new(0)); + let far = Arc::new(AtomicUsize::new(0)); + let (n, f) = (near.clone(), far.clone()); + + let _s_near = node_near + .create_sub::("mixed") + .build_with_shared_callback(move |_m: Arc| { + n.fetch_add(1, Ordering::SeqCst); + })?; + let _s_far = node_far + .create_sub::("mixed") + .build_with_callback(move |_m: RosString| { + f.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node_tx.create_pub::("mixed").build()?; + wait_for_ready(Duration::from_millis(800)); + + publisher.publish_shared(Arc::new(RosString { + data: "both audiences".to_owned(), + }))?; + + // The far one is the control: without it crossing, a zero on the near side + // would be indistinguishable from a broken router. + assert!( + wait_until(|| far.load(Ordering::SeqCst) >= 1), + "the off-session subscriber never received it; this run proves nothing" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!( + near.load(Ordering::SeqCst), + 1, + "the near subscriber received {} copies — the bus and the wire both fired", + near.load(Ordering::SeqCst) + ); + Ok(()) +} + +/// #36 — a sole receiver is handed the message itself, not a shared `Arc`. +/// +/// The proof is that the callback can **mutate** what it receives. A shared +/// `Arc` cannot be mutated at all, so this does not compile against the +/// shared path — which is the point of having a separate one. +#[test] +fn a_sole_receiver_is_given_the_message_to_own() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("owned_node").build()?; + + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let sink = seen.clone(); + let _sub = node + .create_sub::("owned") + .build_with_owned_callback(move |mut msg: RosString| { + // Mutating the received message is the whole property under test. + msg.data.push_str(" — mutated by its owner"); + sink.lock().expect("poisoned").push(msg.data); + })?; + + let publisher = node + .create_pub::("owned") + .with_intra_process_only() + .build()?; + + let took = publisher.publish_owned(RosString { + data: "mine".to_owned(), + })?; + assert_eq!( + took, + Published::Bus(Delivery::Sent(1)), + "the sole owning receiver did not take the message" + ); + + let got = seen.lock().expect("poisoned"); + assert_eq!(got.len(), 1, "the owned callback did not run"); + assert_eq!(got[0], "mine — mutated by its owner"); + Ok(()) +} + +/// #36 defect 1. An owned subscriber's wire half was `|_m| {}`, so every +/// message from off-session — that is, from the entire rest of the ROS graph — +/// was discarded in silence while the subscription still advertised itself. +/// +/// The publisher here is in a *different* context, so the bus cannot carry it +/// and only the wire half can satisfy this test. +#[test] +fn an_owned_subscriber_receives_from_an_off_session_publisher() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx_rx = create_hiroz_context_with_router(&router)?; + let ctx_tx = create_hiroz_context_with_router(&router)?; + let node_rx = ctx_rx.create_node("own_wire_rx").build()?; + let node_tx = ctx_tx.create_node("own_wire_tx").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node_rx + .create_sub::("owned_wire") + .build_with_owned_callback(move |m: RosString| { + assert_eq!(m.data, "from far away"); + h.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node_tx.create_pub::("owned_wire").build()?; + wait_for_ready(Duration::from_millis(800)); + publisher.publish(&RosString { + data: "from far away".to_owned(), + })?; + + assert!( + wait_until(|| hits.load(Ordering::SeqCst) >= 1), + "an owned subscriber discarded a message from another session" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!(hits.load(Ordering::SeqCst), 1, "delivered more than once"); + Ok(()) +} + +/// #36 defect 2. Depth exhaustion and "no subscriber wanted it" both reported +/// zero, and the caller read zero as "fall back to the wire". The wire re-enters +/// the same callback on a zenoh thread, where the depth counter is a fresh +/// thread-local zero, and the publish never returns. +/// +/// The publisher is `Locality::Remote`: the bus-taking path that does NOT use +/// `with_intra_process_only()`, which is what the existing recursion test +/// covers. Its wire half cannot reach this session, so the wire cannot echo and +/// the depth guard is the only thing bounding this. +/// +/// Deliberately not tested with a *plain* publisher: since #134 that takes the +/// wire alone, so a callback republishing onto its own topic is an ordinary +/// topic cycle — the same unbounded echo any ROS 2 node produces by subscribing +/// and publishing to one topic. That is the user's cycle to avoid, not a +/// defect, and no client library prevents it. +#[test] +fn a_self_publishing_callback_does_not_escape_to_the_wire_and_loop() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("loop_node").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + // The publisher is stored erased, so this test does not have to name + // `ZPub`'s serializer parameter. `OnceLock` rather than `Mutex` because + // delivery is synchronous: the callback runs on the publishing thread and + // would re-enter a lock that thread already holds. + type Echo = Arc) + Send + Sync>>>; + let echo: Echo = Arc::new(OnceLock::new()); + let echo_cb = echo.clone(); + let _sub = node + .create_sub::("cycle") + .build_with_shared_callback(move |m: Arc| { + let seen = h.fetch_add(1, Ordering::SeqCst); + // Stop echoing well above the depth guard's bound but well below + // forever. Without this the unbounded case never terminates, and a + // detector that hangs tells you less than one that fails: the + // timeout does not say which property broke. + if seen < ECHO_CAP + && let Some(publish) = echo_cb.get() + { + publish(m); + } + })?; + + let publisher = node + .create_pub::("cycle") + .with_locality(zenoh::sample::Locality::Remote) + .build()?; + let _ = echo.set(Arc::new(move |m: Arc| { + let _ = publisher.publish_shared(m); + })); + wait_for_ready(Duration::from_millis(500)); + + // Publish on a worker thread behind a watchdog. Escaping to the wire does + // not merely loop: the wire callback runs on a zenoh runtime thread and + // publishes from inside it, which blocks on that runtime — a re-entrancy + // deadlock. A count assertion cannot see that, because the count never gets + // to climb. Only a timeout can, so the timeout is made explicit here rather + // than left to the CI runner, where it would report as "the suite hung" + // without naming the property. + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let counter = hits.clone(); + let publish = echo.get().expect("just set").clone(); + thread::spawn(move || { + publish(Arc::new(RosString { + data: "round and round".to_owned(), + })); + thread::sleep(Duration::from_millis(400)); + let settled = counter.load(Ordering::SeqCst); + thread::sleep(Duration::from_millis(600)); + let _ = done_tx.send((settled, counter.load(Ordering::SeqCst))); + }); + + let (settled, later) = done_rx.recv_timeout(Duration::from_secs(10)).expect( + "publish never returned: the depth guard escaped to the wire, \ + which re-enters on a zenoh thread and deadlocks on its runtime", + ); + assert!( + later <= 32, + "{later} deliveries from one publish (settled at {settled}): the guard \ + escaped to the wire and re-entered at depth zero" + ); + assert_eq!( + settled, later, + "delivery is still growing ({settled} -> {later}) after it should have stopped" + ); + Ok(()) +} + +/// #36 defect 3. `bus_can_serve_everyone` never consulted the publisher's own +/// locality. With `Locality::Remote` the wire half cannot reach this session, so +/// taking the wire alone left the same-session subscriber with nothing. +/// +/// The far subscriber is the positive control: without it crossing, a zero on +/// the near side would be indistinguishable from a broken router. +#[test] +fn a_remote_locality_publisher_still_reaches_a_same_session_subscriber() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx_tx = create_hiroz_context_with_router(&router)?; + let ctx_far = create_hiroz_context_with_router(&router)?; + let node_tx = ctx_tx.create_node("rem_tx").build()?; + let node_near = ctx_tx.create_node("rem_near").build()?; + let node_far = ctx_far.create_node("rem_far").build()?; + + let near = Arc::new(AtomicUsize::new(0)); + let far = Arc::new(AtomicUsize::new(0)); + let (n, f) = (near.clone(), far.clone()); + + let _s_near = node_near + .create_sub::("split") + .build_with_shared_callback(move |_m: Arc| { + n.fetch_add(1, Ordering::SeqCst); + })?; + let _s_far = node_far + .create_sub::("split") + .build_with_callback(move |_m: RosString| { + f.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node_tx + .create_pub::("split") + .with_locality(zenoh::sample::Locality::Remote) + .build()?; + wait_for_ready(Duration::from_millis(800)); + + let delivered = publisher.publish_shared(Arc::new(RosString { + data: "both, disjointly".to_owned(), + }))?; + assert_eq!( + delivered, + Published::BusAndWire(Delivery::Sent(1)), + "the bus did not carry it to the near subscriber, or skipped the wire" + ); + + assert!( + wait_until(|| far.load(Ordering::SeqCst) >= 1), + "the off-session subscriber never received it; this run proves nothing" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!( + near.load(Ordering::SeqCst), + 1, + "near subscriber count wrong" + ); + assert_eq!(far.load(Ordering::SeqCst), 1, "far subscriber count wrong"); + Ok(()) +} + +/// #133. TRANSIENT_LOCAL durability lives in the wire publisher's cache, and an +/// intra-process-only publisher has no wire. Serving the bus would let a +/// late-joining subscriber be handed a history this message is missing from — +/// wrong rather than absent — so the call is refused instead. +#[test] +fn transient_local_plus_intra_process_only_is_refused() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("tl_node").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node + .create_sub::("tl_topic") + .build_with_shared_callback(move |_m: Arc| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("tl_topic") + .with_qos(hiroz::qos::QosProfile { + durability: hiroz::qos::QosDurability::TransientLocal, + ..Default::default() + }) + .with_intra_process_only() + .build()?; + wait_for_ready(Duration::from_millis(500)); + + let result = publisher.publish_shared(Arc::new(RosString { + data: "durable, allegedly".to_owned(), + })); + assert!( + result.is_err(), + "a transient-local publisher served the bus, which has no durability cache" + ); + // And it did not deliver it anyway: a refusal that still delivers is worse + // than either outcome on its own. + thread::sleep(Duration::from_millis(200)); + assert_eq!( + hits.load(Ordering::SeqCst), + 0, + "the call was refused but the message was delivered regardless" + ); + Ok(()) +} + +/// The `Some(second)` branch of `Channel::publish` has no other coverage: every +/// other test in this file uses exactly one shared subscriber, so a stub that +/// called only the first and returned `Sent(1)` would pass the whole suite. +/// +/// `Arc::ptr_eq` on *every* receiver is the point. Delivering a clone to the +/// second and third is what the branch exists to avoid. +#[test] +fn every_shared_subscriber_receives_the_same_allocation() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("fanout").build()?; + + let seen: Arc>>> = Arc::new(Mutex::new(Vec::new())); + let mut subs = Vec::new(); + for _ in 0..3 { + let sink = seen.clone(); + subs.push( + node.create_sub::("fanout") + .build_with_shared_callback(move |m: Arc| { + sink.lock().expect("poisoned").push(m); + })?, + ); + } + + let publisher = node + .create_pub::("fanout") + .with_intra_process_only() + .build()?; + let sent = Arc::new(RosString { + data: "one allocation, three readers".to_owned(), + }); + let delivered = publisher.publish_shared(sent.clone())?; + assert_eq!( + delivered, + Published::Bus(Delivery::Sent(3)), + "not every subscriber was served" + ); + + let got = seen.lock().expect("poisoned"); + assert_eq!(got.len(), 3, "expected three deliveries"); + for (i, g) in got.iter().enumerate() { + assert!( + Arc::ptr_eq(&sent, g), + "receiver {i} was handed a different allocation: the fan-out branch copies" + ); + } + Ok(()) +} + +/// #133 reached through the sibling method. `publish_shared` refuses a +/// transient-local intra-process-only publisher; `publish_owned` took the bus +/// around that check and returned `Ok(1)`. +#[test] +fn transient_local_plus_intra_process_only_is_refused_for_owned_too() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("tl_owned").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node + .create_sub::("tl_owned_topic") + .build_with_owned_callback(move |_m: RosString| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("tl_owned_topic") + .with_qos(hiroz::qos::QosProfile { + durability: hiroz::qos::QosDurability::TransientLocal, + ..Default::default() + }) + .with_intra_process_only() + .build()?; + wait_for_ready(Duration::from_millis(500)); + + let result = publisher.publish_owned(RosString { + data: "durable, allegedly".to_owned(), + }); + assert!( + result.is_err(), + "publish_owned served the bus for a transient-local publisher" + ); + thread::sleep(Duration::from_millis(200)); + assert_eq!( + hits.load(Ordering::SeqCst), + 0, + "refused and delivered anyway" + ); + Ok(()) +} + +/// A panicking subscriber must not censor its siblings or kill the publisher. +/// +/// Bus delivery is synchronous on the publishing thread, so without isolation a +/// panic unwinds out of `publish_shared` into the application and every +/// subscriber later in the snapshot is skipped. Delivery order is snapshot +/// order, so which siblings are lost varies between runs. +/// +/// The wire path gives this isolation for free — one panicking callback kills +/// one zenoh task. This test exists so the bus does not regress against it. +#[test] +#[serial_test::serial(panic_hook)] +fn a_panicking_subscriber_does_not_stop_delivery_to_the_others() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("panicky").build()?; + + let before = Arc::new(AtomicUsize::new(0)); + let after = Arc::new(AtomicUsize::new(0)); + let (b, a) = (before.clone(), after.clone()); + + // The panicking subscriber is registered FIRST, deliberately. Delivery + // walks the snapshot in registration order and the fan-out branch invokes + // the first entry on its own line, so a panic anywhere later leaves that + // line untested — which is how an earlier version of this test passed + // against the isolation being removed. + let _s1 = node + .create_sub::("panicky") + .build_with_shared_callback(move |_m: Arc| { + panic!("this subscriber is deliberately broken"); + })?; + let _s2 = node + .create_sub::("panicky") + .build_with_shared_callback(move |_m: Arc| { + b.fetch_add(1, Ordering::SeqCst); + })?; + let _s3 = node + .create_sub::("panicky") + .build_with_shared_callback(move |_m: Arc| { + a.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("panicky") + .with_intra_process_only() + .build()?; + + // The panic is reported by the default hook; silence it so the test output + // is readable, and restore the hook afterwards. + let prev = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let delivered = publisher.publish_shared(Arc::new(RosString { + data: "one of you will panic".to_owned(), + })); + std::panic::set_hook(prev); + + // The publisher returned at all: the panic did not unwind into this thread. + let delivered = delivered.expect("the panic escaped into the publishing thread"); + assert_eq!( + delivered, + Published::Bus(Delivery::Sent(2)), + "three subscribers were called and two returned. The count is deliveries, \ + not invocations — the assertions on the two counters below are what \ + prove the panicking one did not stop its siblings." + ); + assert_eq!( + before.load(Ordering::SeqCst), + 1, + "the subscriber after the panicking one was skipped" + ); + assert_eq!( + after.load(Ordering::SeqCst), + 1, + "the subscriber after the panicking one was skipped: one bad callback \ + censored its siblings" + ); + Ok(()) +} + +/// The sole-subscriber branch of `Channel::publish` is a different call site +/// from the fan-out branch, and a panic there has nowhere else to go: it would +/// unwind straight out of `publish_shared` into the caller. +#[test] +#[serial_test::serial(panic_hook)] +fn a_panicking_sole_subscriber_does_not_reach_the_publisher() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("panicky_solo").build()?; + + let _sub = node + .create_sub::("panicky_solo") + .build_with_shared_callback(move |_m: Arc| { + panic!("the only subscriber, and it is broken"); + })?; + let publisher = node + .create_pub::("panicky_solo") + .with_intra_process_only() + .build()?; + + let prev = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let delivered = publisher.publish_shared(Arc::new(RosString { + data: "boom".to_owned(), + })); + std::panic::set_hook(prev); + + let delivered = delivered.expect("the panic escaped into the publishing thread"); + assert_eq!( + delivered, + Published::Bus(Delivery::NoTaker), + "the sole subscriber panicked, so nothing was delivered. Sent(1) would \ + report a message as landed when none was: Delivery::Sent counts \ + subscribers that returned, not subscribers that were called." + ); + Ok(()) +} + +/// B1. `publish_owned` on a `Locality::Remote` publisher must reach the wire. +/// +/// This is the detector that did not exist. Every other `publish_owned` test +/// uses `with_intra_process_only()`, so none of them could see that the Remote +/// arm took the bus and returned — losing the message to every off-session +/// subscriber, silently, while `publish_shared` in the identical configuration +/// served both. +/// +/// The far assertion is the point. The near one passes against the defect. +#[test] +fn publish_owned_on_a_remote_publisher_still_reaches_the_wire() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx_tx = create_hiroz_context_with_router(&router)?; + let ctx_far = create_hiroz_context_with_router(&router)?; + let node_tx = ctx_tx.create_node("owned_rem_tx").build()?; + let node_near = ctx_tx.create_node("owned_rem_near").build()?; + let node_far = ctx_far.create_node("owned_rem_far").build()?; + + let near = Arc::new(AtomicUsize::new(0)); + let far = Arc::new(AtomicUsize::new(0)); + let (n, f) = (near.clone(), far.clone()); + + let _s_near = node_near + .create_sub::("owned_split") + .build_with_owned_callback(move |_m: RosString| { + n.fetch_add(1, Ordering::SeqCst); + })?; + let _s_far = node_far + .create_sub::("owned_split") + .build_with_callback(move |_m: RosString| { + f.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node_tx + .create_pub::("owned_split") + .with_locality(zenoh::sample::Locality::Remote) + .build()?; + wait_for_ready(Duration::from_millis(800)); + + let outcome = publisher.publish_owned(RosString { + data: "must reach both".to_owned(), + })?; + + // Shared, not moved: the wire half needs the value to serialize it, so a + // Remote publisher cannot give it away. + assert!( + matches!(outcome, Published::BusAndWire(_)), + "a Remote publisher must run both routes, got {outcome:?}" + ); + + assert!( + wait_until(|| far.load(Ordering::SeqCst) >= 1), + "the off-session subscriber never received it — this is B1" + ); + thread::sleep(Duration::from_millis(300)); + assert_eq!( + near.load(Ordering::SeqCst), + 1, + "near subscriber count wrong" + ); + assert_eq!(far.load(Ordering::SeqCst), 1, "far subscriber count wrong"); + Ok(()) +} + +/// B2. The durability refusal permits a `Locality::Remote` publisher because +/// the wire still populates its cache. That justification has to hold on the +/// owned path too, or a late joiner is served a history this message is +/// missing from. +#[test] +fn transient_local_plus_remote_locality_is_allowed_on_the_owned_path() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("tl_rem_node").build()?; + + let publisher = node + .create_pub::("tl_rem_topic") + .with_qos(hiroz::qos::QosProfile { + durability: hiroz::qos::QosDurability::TransientLocal, + ..Default::default() + }) + .with_locality(zenoh::sample::Locality::Remote) + .build()?; + wait_for_ready(Duration::from_millis(300)); + + // Permitted, because the wire runs and fills the cache. The companion test + // above is what proves the wire actually runs; without it this assertion + // would be satisfied by a publisher that silently dropped the message. + let outcome = publisher.publish_owned(RosString { + data: "durable".to_owned(), + })?; + assert!( + matches!(outcome, Published::BusAndWire(_) | Published::Wire), + "a durable Remote publisher must still reach the wire, got {outcome:?}" + ); + Ok(()) +} + +/// B3. `Published` must distinguish the three ways nothing was delivered +/// locally, because a caller may fall back to the wire on one of them and must +/// not on another. +/// +/// A count cannot carry this: `NoTaker`, `DepthExceeded` and "there is no bus +/// on this publisher" were all zero. +#[test] +fn published_says_which_routes_ran_and_why_nothing_was_taken() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("outcome_node").build()?; + + // 1. No assertion from the caller: the wire alone, and the bus is not + // consulted at all. Not `Bus(NoTaker)` — nothing asked the bus. + let plain = node.create_pub::("outcome_plain").build()?; + let out = plain.publish_shared(Arc::new(RosString { + data: "wire".to_owned(), + }))?; + assert_eq!(out, Published::Wire, "a plain publisher must report Wire"); + + // 2. Bus asserted, nobody listening. Distinct from the case above: here the + // bus WAS asked and had no taker. + let local = node + .create_pub::("outcome_local") + .with_intra_process_only() + .build()?; + let out = local.publish_shared(Arc::new(RosString { + data: "nobody".to_owned(), + }))?; + assert_eq!( + out, + Published::Bus(Delivery::NoTaker), + "an empty bus must report NoTaker, not Wire" + ); + + // 3. Bus asserted, one subscriber. + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node + .create_sub::("outcome_taken") + .build_with_shared_callback(move |_m: Arc| { + h.fetch_add(1, Ordering::SeqCst); + })?; + let taken = node + .create_pub::("outcome_taken") + .with_intra_process_only() + .build()?; + let out = taken.publish_shared(Arc::new(RosString { + data: "taken".to_owned(), + }))?; + assert_eq!(out, Published::Bus(Delivery::Sent(1))); + assert_eq!(hits.load(Ordering::SeqCst), 1); + Ok(()) +} + +/// B3, the half a count could never express: a message refused at +/// `MAX_DELIVERY_DEPTH` is dropped, and must not be reported as delivered. +/// +/// Before this, the bus returned `Ok(())` for both outcomes and `ZPub` mapped +/// it to `Ok(1)` — a dropped message counted as one receiver taking it. A +/// caller that retries on a low count would have retried forever on the one +/// outcome where retrying re-enters the same callback. +#[test] +fn a_depth_refusal_is_reported_as_dropped_not_delivered() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("depth_outcome").build()?; + + let outcomes: Arc>> = Arc::new(Mutex::new(Vec::new())); + let publisher: Arc> = Arc::new(OnceLock::new()); + + let p = publisher.clone(); + let o = outcomes.clone(); + let _sub = node + .create_sub::("depth_outcome") + .build_with_owned_callback(move |m: RosString| { + // Republish onto the same topic: each delivery nests one deeper. + if let Some(send) = p.get() + && let Ok(out) = send(m) + { + o.lock().expect("outcome lock").push(out); + } + })?; + + let pubr = node + .create_pub::("depth_outcome") + .with_intra_process_only() + .build()?; + let pubr = Arc::new(pubr); + let p2 = pubr.clone(); + publisher + .set(Box::new(move |m: RosString| p2.publish_owned(m))) + .map_err(|_| "publisher already set") + .expect("set once"); + + pubr.publish_owned(RosString { + data: "recurse".to_owned(), + })?; + + let seen = outcomes.lock().expect("outcome lock"); + assert!( + seen.iter() + .any(|o| matches!(o, Published::Bus(Delivery::DepthExceeded))), + "the depth refusal was never reported; outcomes were {seen:?}" + ); + // A plain global bound, not a per-element predicate. The previous form put + // the loop-invariant `seen.len()` inside `any(..)`, where the guard's own + // bound made it false for every element — so the assertion was `!false` for + // all inputs and could never fail. Without this, "refused once at depth 8" + // and "refused once at depth 800" are indistinguishable. + assert!( + seen.len() <= hiroz::local_bus::MAX_DELIVERY_DEPTH as usize + 2, + "delivery went {} deep; the depth bound is {}", + seen.len(), + hiroz::local_bus::MAX_DELIVERY_DEPTH + ); + Ok(()) +} + +/// S9. `publish_shared` and `publish_owned` must agree about where a message +/// goes. They cannot, unless they read the same decision. +/// +/// They did not. Each tested the conditions itself, in the opposite order, so a +/// publisher carrying both assertions routed one way through one method and the +/// other way through the other — the same drift that produced B1, in a +/// configuration nothing exercised. `with_intra_process_only()` wins, because it +/// is the stronger claim: this publisher has no wire. +#[test] +fn both_publish_methods_agree_when_the_assertions_conflict() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("conflict_node").build()?; + + let publisher = node + .create_pub::("conflict_topic") + .with_intra_process_only() + .with_locality(zenoh::sample::Locality::Remote) + .build()?; + wait_for_ready(Duration::from_millis(300)); + + let shared = publisher.publish_shared(Arc::new(RosString { + data: "shared".to_owned(), + }))?; + let moved = publisher.publish_owned(RosString { + data: "moved".to_owned(), + })?; + + // Nobody is subscribed, so both report the bus having no taker. The point is + // that neither reached the wire: a `BusAndWire` from either is the drift. + assert_eq!( + shared, + Published::Bus(Delivery::NoTaker), + "publish_shared took the wrong route" + ); + assert_eq!( + moved, + Published::Bus(Delivery::NoTaker), + "publish_owned disagreed with publish_shared" + ); + Ok(()) +} + +/// An owned subscriber must not be starved in silence. +/// +/// `Channel::publish` filters on `is_shared()`, so an owned subscriber is +/// invisible to the shared path — and an `intra_process_only` publisher has no +/// wire behind the bus. Before this, the message vanished and the caller was +/// told `Ok(Bus(NoTaker))`: a registered receiver of the right type existed and +/// got nothing, with one `debug!` line as the only trace. +/// +/// It cannot be repaired by serving a clone, because `ZMessage` is not `Clone`. +/// So the publisher refuses instead, the way `refuse_durable_bus` refuses a +/// durable publisher rather than quietly violating its contract. +#[test] +fn publish_shared_refuses_when_only_owned_subscribers_are_listening() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("owned_starve").build()?; + + let hits = Arc::new(AtomicUsize::new(0)); + let h = hits.clone(); + let _sub = node + .create_sub::("owned_starve") + .build_with_owned_callback(move |_m: RosString| { + h.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("owned_starve") + .with_intra_process_only() + .build()?; + wait_for_ready(Duration::from_millis(300)); + + let outcome = publisher.publish_shared(Arc::new(RosString { + data: "nobody shared is listening".to_owned(), + })); + + assert!( + outcome.is_err(), + "an owned subscriber was registered and could not be served, and there is \ + no wire: this must refuse, not report success. Got {outcome:?}" + ); + // The control: it really was unserviceable, not merely refused. + assert_eq!( + hits.load(Ordering::SeqCst), + 0, + "the owned subscriber cannot be served by a shared publish" + ); + Ok(()) +} + +/// The same starvation reached through `publish_owned`, which is the likelier +/// route to it: two owning subscribers means nothing can be given away, so it +/// falls back to sharing — into the case above. +#[test] +fn publish_owned_refuses_when_two_owned_subscribers_cannot_be_served() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("owned_two").build()?; + + let a = Arc::new(AtomicUsize::new(0)); + let b = Arc::new(AtomicUsize::new(0)); + let (ca, cb) = (a.clone(), b.clone()); + let _s1 = node + .create_sub::("owned_two") + .build_with_owned_callback(move |_m: RosString| { + ca.fetch_add(1, Ordering::SeqCst); + })?; + let _s2 = node + .create_sub::("owned_two") + .build_with_owned_callback(move |_m: RosString| { + cb.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("owned_two") + .with_intra_process_only() + .build()?; + wait_for_ready(Duration::from_millis(300)); + + let outcome = publisher.publish_owned(RosString { + data: "two takers, nothing to give".to_owned(), + }); + + assert!( + outcome.is_err(), + "two owning subscribers cannot both be given the value, and neither can be \ + served by the shared fallback: this must refuse. Got {outcome:?}" + ); + assert_eq!( + (a.load(Ordering::SeqCst), b.load(Ordering::SeqCst)), + (0, 0), + "neither owned subscriber is serviceable here" + ); + Ok(()) +} + +/// A wire failure must return **before** any local subscriber has run. +/// +/// The previous version of this test published successfully and asserted both +/// routes ran. That passes equally well with the bus first, so it did not +/// protect the property it was named for. The property is only observable in +/// the failure case: bus delivery is synchronous, so if the bus ran first, a +/// caller that retries after an `Err` delivers to every local subscriber twice +/// — and `Result` has no partial-success value to warn them. +/// +/// The session is closed to make the wire fail deterministically. +#[test] +fn a_wire_failure_returns_before_any_local_subscriber_runs() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("order_fail").build()?; + + let near = Arc::new(AtomicUsize::new(0)); + let n = near.clone(); + let _sub = node + .create_sub::("order_fail_topic") + .build_with_shared_callback(move |_m: Arc| { + n.fetch_add(1, Ordering::SeqCst); + })?; + + let publisher = node + .create_pub::("order_fail_topic") + .with_locality(zenoh::sample::Locality::Remote) + .build()?; + wait_for_ready(Duration::from_millis(300)); + + // The control: while the session is alive, both routes run. Without this a + // broken setup would satisfy the assertion below for the wrong reason. + let ok = publisher.publish_shared(Arc::new(RosString { + data: "healthy".to_owned(), + }))?; + assert!( + matches!(ok, Published::BusAndWire(Delivery::Sent(1))), + "control: a healthy publish must run both routes, got {ok:?}" + ); + assert_eq!( + near.load(Ordering::SeqCst), + 1, + "control: the bus must deliver" + ); + + ctx.shutdown()?; + let after_control = near.load(Ordering::SeqCst); + + let failed = publisher.publish_shared(Arc::new(RosString { + data: "the wire is gone".to_owned(), + })); + assert!( + failed.is_err(), + "the session is closed, so the wire half must fail; got {failed:?}" + ); + assert_eq!( + near.load(Ordering::SeqCst), + after_control, + "the wire failed, so no local subscriber may have run. Running the bus \ + first and failing afterwards leaves the caller unable to retry safely." + ); + Ok(()) +} + +/// A channel nobody holds any more is reclaimed, so the registry does not grow +/// for the life of the process. +/// +/// `channel()` runs in **every** `ZPubBuilder::build()`, so this is not one +/// entry per `ZContext` — a process publishing on dynamically-named topics +/// accumulated one entry per topic, permanently, long after every publisher was +/// dropped. Reclamation is safe only because an `Arc::strong_count` of one +/// proves no `ZPub` or `ZSub` can still reach the channel, so a later builder +/// cannot be split from a live endpoint. +#[test] +#[serial_test::serial(registry)] +fn a_channel_no_endpoint_holds_is_reclaimed() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_router(&router)?; + let node = ctx.create_node("reclaim").build()?; + + // Serialised: total_channels() is process-global and sibling tests create + // channels of their own, so an unsynchronised delta is not attributable. + let before = hiroz::local_bus::total_channels(); + for i in 0..24 { + let p = node + .create_pub::(&format!("ephemeral_{i}")) + .build()?; + drop(p); + } + let after_ephemeral = hiroz::local_bus::total_channels(); + // saturating: reclamation triggered by another test can legitimately make + // the total fall, and an unsigned underflow would panic on a healthy run. + assert!( + after_ephemeral.saturating_sub(before) <= 4, + "24 publishers were created and dropped; the registry grew by {} channels. \ + It should reclaim the ones no endpoint holds.", + after_ephemeral.saturating_sub(before) + ); + + // The control. Without it this test passes equally well against a registry + // that reclaims indiscriminately — which would split a live publisher from + // a subscriber created afterwards, turning a leak into lost messages. + let held = node.create_pub::("held_topic").build()?; + let with_held = hiroz::local_bus::total_channels(); + let _forces_a_reclaim_pass = node.create_pub::("spacer_topic").build()?; + assert!( + hiroz::local_bus::total_channels() >= with_held, + "a channel a live publisher still holds must survive a reclamation pass" + ); + drop(held); + Ok(()) +} diff --git a/crates/hiroz-tests/tests/publisher_locality.rs b/crates/hiroz-tests/tests/publisher_locality.rs new file mode 100644 index 000000000..abda0e21a --- /dev/null +++ b/crates/hiroz-tests/tests/publisher_locality.rs @@ -0,0 +1,146 @@ +//! `ZPubBuilder::with_locality` — does the restriction actually reach the wire? +//! +//! `Locality::SessionLocal` is the intra-process fast path. Zenoh's +//! `resolve_put` skips `primitives.send_push_consume` entirely for it, so the +//! samples never leave the session: no link, no wire encode, no shared memory. +//! Subscribers in the same session still get them, through the local callback +//! list. +//! +//! # What makes this a detector rather than a green tick +//! +//! "The other context received nothing" is worthless on its own — a broken +//! router, a wrong topic or a too-short deadline all produce the same zero. So +//! every scenario publishes on TWO topics from the SAME context: +//! +//! | topic | publisher locality | same context | other context | +//! |---|---|---|---| +//! | `local_only` | `SessionLocal` | must receive | must receive NOTHING | +//! | `control` | default (`Any`) | must receive | **must receive** | +//! +//! The control row is the point. It proves the router routes, the topics match +//! and the deadline is long enough, using the same processes and the same +//! deadline as the row under test. Without it a zero is unknown, not a pass. +//! +//! Revert `with_locality` (or pass `Locality::Any`) and `local_only` starts +//! arriving in the other context, which fails the assertion. + +mod common; + +use std::{ + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + thread, + time::{Duration, Instant}, +}; + +use common::*; +use hiroz::{Builder, Locality}; +use hiroz_msgs::std_msgs::String as RosString; + +const MESSAGES: usize = 5; +/// Generous relative to loopback delivery. It bounds the "received nothing" +/// assertions, so it has to be long enough that a slow-but-working path is not +/// mistaken for a blocked one. +const DELIVERY_DEADLINE: Duration = Duration::from_secs(5); + +/// A string subscriber, spelled once so the signature below stays readable. +/// +/// `ZSub` carries a serializer parameter, so writing the full type at the +/// return position is what trips `clippy::type_complexity`. +type StringSub = hiroz::pubsub::ZSub>; + +fn counting_sub( + node: &hiroz::node::ZNode, + topic: &str, +) -> hiroz::Result<(StringSub, Arc)> { + let count = Arc::new(AtomicUsize::new(0)); + let c = count.clone(); + let sub = node + .create_sub::(topic) + .build_with_callback(move |_msg: RosString| { + c.fetch_add(1, Ordering::SeqCst); + })?; + Ok((sub, count)) +} + +/// Poll until `f` holds or the deadline passes. Returns whether it held. +fn wait_until(f: impl Fn() -> bool) -> bool { + let start = Instant::now(); + while start.elapsed() < DELIVERY_DEADLINE { + if f() { + return true; + } + thread::sleep(Duration::from_millis(20)); + } + f() +} + +#[test] +fn session_local_publisher_is_invisible_to_another_context() -> hiroz::Result<()> { + let router = TestRouter::new(); + let ctx_publisher = create_hiroz_context_with_router(&router)?; + let ctx_other = create_hiroz_context_with_router(&router)?; + + let node_pub = ctx_publisher.create_node("locality_pub").build()?; + let node_same = ctx_publisher.create_node("locality_same").build()?; + let node_other = ctx_other.create_node("locality_other").build()?; + + let (_s1, same_local) = counting_sub(&node_same, "local_only")?; + let (_s2, other_local) = counting_sub(&node_other, "local_only")?; + let (_s3, same_control) = counting_sub(&node_same, "control")?; + let (_s4, other_control) = counting_sub(&node_other, "control")?; + + let pub_local = node_pub + .create_pub::("local_only") + .with_locality(Locality::SessionLocal) + .build()?; + let pub_control = node_pub.create_pub::("control").build()?; + + // Let the router propagate the four subscriber declarations before the first + // put, or the control row can lose messages for a reason unrelated to + // locality and take the test red for the wrong cause. + wait_for_ready(Duration::from_millis(500)); + + for i in 0..MESSAGES { + let msg = RosString { + data: format!("msg-{i}"), + }; + pub_local.publish(&msg)?; + pub_control.publish(&msg)?; + } + + // The control must arrive in BOTH contexts. Assert it first: if it fails, + // the environment is broken and the SessionLocal assertions below would be + // meaningless rather than informative. + assert!( + wait_until(|| same_control.load(Ordering::SeqCst) >= MESSAGES), + "control publisher did not reach the same context: {} of {MESSAGES}", + same_control.load(Ordering::SeqCst) + ); + assert!( + wait_until(|| other_control.load(Ordering::SeqCst) >= MESSAGES), + "control publisher did not reach the other context: {} of {MESSAGES}. \ + The router is not routing, so this run proves nothing about locality.", + other_control.load(Ordering::SeqCst) + ); + + // Same session: SessionLocal still delivers. + assert!( + wait_until(|| same_local.load(Ordering::SeqCst) >= MESSAGES), + "SessionLocal publisher did not reach a subscriber in its own session: {} of {MESSAGES}", + same_local.load(Ordering::SeqCst) + ); + + // Other session: it must have delivered nothing. The control above already + // arrived over the same router within the same deadline, so a zero here is + // the restriction working, not a slow path. + assert_eq!( + other_local.load(Ordering::SeqCst), + 0, + "SessionLocal publisher leaked to another context — the restriction is not applied" + ); + + Ok(()) +} diff --git a/crates/hiroz/Cargo.toml b/crates/hiroz/Cargo.toml index f8405d3c9..23534ecb6 100644 --- a/crates/hiroz/Cargo.toml +++ b/crates/hiroz/Cargo.toml @@ -30,6 +30,9 @@ zenoh = { workspace = true, default-features = false, features = [ ] } zenoh-ext = { workspace = true } zenoh-buffers = { workspace = true } +# Snapshot for the intra-process subscriber list: the publish path reads it +# without a lock and without cloning callbacks. Already in the tree via zenoh. +arc-swap = "1.7" strum = { workspace = true, features = ["derive"] } sha2 = { workspace = true } parking_lot = { workspace = true } diff --git a/crates/hiroz/src/dynamic/tests/pubsub_tests.rs b/crates/hiroz/src/dynamic/tests/pubsub_tests.rs index 6cbc2c393..1f7150149 100644 --- a/crates/hiroz/src/dynamic/tests/pubsub_tests.rs +++ b/crates/hiroz/src/dynamic/tests/pubsub_tests.rs @@ -184,6 +184,8 @@ fn test_zpub_builder_with_dyn_schema() { shm_config: None, dyn_schema: None, encoding: None, + locality: None, + intra_process_only: false, _phantom_data: PhantomData, }; @@ -226,6 +228,8 @@ fn test_zpub_builder_with_serdes_preserves_schema() { shm_config: None, dyn_schema: Some(schema.clone()), encoding: None, + locality: None, + intra_process_only: false, _phantom_data: PhantomData, }; diff --git a/crates/hiroz/src/lib.rs b/crates/hiroz/src/lib.rs index 7cf631043..5237b12bd 100644 --- a/crates/hiroz/src/lib.rs +++ b/crates/hiroz/src/lib.rs @@ -65,6 +65,8 @@ pub mod ffi; pub mod graph; /// ROS 2 lifecycle node support (state machine, lifecycle publisher). pub mod lifecycle; +/// Intra-process message bus (prototype) — `Arc` delivery without serialization. +pub mod local_bus; /// Typed message wrappers and helpers. pub mod msg; /// ROS 2 node creation and management. @@ -110,6 +112,10 @@ pub use hiroz_derive::MessageTypeInfo; pub use ros_msg::{ActionTypeInfo, MessageTypeInfo, ServiceTypeInfo, WithTypeInfo}; pub use zbuf::ZBuf; pub use zenoh::Result; +/// Re-exported so callers of `with_locality` do not need a direct `zenoh` +/// dependency — and so they cannot pick a different `zenoh` version than the +/// one hiroz links. +pub use zenoh::sample::Locality; /// Builds a configured object, consuming the builder. /// diff --git a/crates/hiroz/src/local_bus.rs b/crates/hiroz/src/local_bus.rs new file mode 100644 index 000000000..83c76fb1a --- /dev/null +++ b/crates/hiroz/src/local_bus.rs @@ -0,0 +1,638 @@ +//! Intra-process message bus — **prototype**. +//! +//! Zenoh's payload is bytes by definition, so every hiroz message is CDR-encoded +//! on publish and decoded on receive even when both endpoints live in the same +//! process. [`Locality::SessionLocal`](zenoh::sample::Locality) removes the +//! transport underneath that, but not the encoding. +//! +//! This module removes the encoding, for the one case where it is safe to: the +//! publisher and the subscriber are in the same zenoh session and agree on the +//! exact same concrete Rust type. The publisher hands out an `Arc` and every +//! matching local subscriber gets a refcount bump. Nothing is serialized and the +//! payload is never copied. +//! +//! # Why this cannot be the default +//! +//! Bytes are what buy ROS its late binding: a topic is a name, and the peer is +//! discovered at run time and may be another process, another language, another +//! ROS version or another host. An `Arc` survives none of that. So this is an +//! opt-in fast path beside the wire path, never a replacement for it — the same +//! shape as rclcpp's intra-process comm. +//! +//! # Scope +//! +//! | | here | not here | +//! |---|---|---| +//! | audience | every same-session subscriber registered on this channel | subscribers reached only over the wire | +//! | type check | exact [`core::any::TypeId`] | any structural or version-tolerant match | +//! | mutability | shared `Arc`, or moved to a sole receiver | a receiver mutating a payload others still hold | +//! | choosing the path | the caller asserts the audience | inferring it | +//! +//! The last row is the one that has been tried and withdrawn. A publisher does +//! not ask whether anyone remote is listening, because the question has no +//! answer: a plain zenoh subscriber declares no ROS liveliness token, so the +//! graph cannot see it and there is no count to subtract our own from. Taking +//! the bus on a wrong inference loses that subscriber's messages silently. So +//! the caller says, with `with_intra_process_only()` or a `Locality`, and the +//! bus is taken only when that assertion makes the wire redundant. +//! +//! # Keying, and why a publisher resolves it once +//! +//! Channels are keyed by `(session zid, topic key expression)`. The zid is what +//! makes "same session" true rather than merely "same process" — two +//! [`ZContext`](crate::context::ZContext)s in one process open two sessions and +//! must not see each other's traffic. Both `ZPub` and `ZSub` already hold an +//! `Arc`, so this needs no plumbing through the node tree. +//! +//! A publisher takes its [`crate::local_bus::Channel`] handle when it is built and never touches +//! the registry again. Resolving per message would mean hashing a +//! fully-qualified ROS key expression on every publish, which at small payloads +//! is a visible share of the whole path. +//! +//! # Locking +//! +//! Callbacks are **never** invoked with the registry lock held. A subscriber +//! callback commonly publishes — that is exactly what the pong side of a +//! ping/pong does — and re-entering the bus under its own read guard is the +//! deadlock this workspace has already fixed three times elsewhere. The list is +//! cloned out, the guard dropped, and only then are the callbacks called. + +use std::{ + any::{Any, TypeId}, + cell::Cell, + collections::HashMap, + sync::{ + Arc, LazyLock, RwLock, + atomic::{AtomicU64, Ordering}, + }, +}; + +// Only the debug-only re-raise below reads this, so the import is gated too. +#[cfg(debug_assertions)] +use crate::reentrancy::ReentrancyViolation; + +use arc_swap::ArcSwap; + +use tracing::debug; +use zenoh::session::ZenohId; + +/// How deep a chain of callback-driven publishes may go before the bus refuses. +/// +/// Delivery runs inline on the publishing thread, so a callback that publishes +/// re-enters `Channel::publish` on the same stack. A pong that publishes to a +/// *different* topic is the motivating case and terminates. A callback that +/// publishes to its **own** topic does not: on the wire that loop passes through +/// zenoh's queues and shows up as an endless stream of messages, but here it is +/// direct recursion and ends in a stack overflow. +/// +/// Eight is chosen to be far above any legitimate chain — a pipeline of eight +/// nodes each publishing from the previous one's callback, on one thread — and +/// far below the depth at which the stack is in danger. +/// Public so a test can assert the exact bound rather than a loose ceiling: +/// a hand-copied constant lets a change to this value slip past unnoticed. +/// +/// # What this does not bound +/// +/// It bounds the **stack**, on **one thread**. Two things escape it, and both +/// are inherent rather than oversights: +/// +/// - **A callback that spawns a thread and publishes there** starts at depth +/// zero, because the counter is thread-local. Unbounded recursion then +/// exhausts threads rather than the stack, and this guard never trips. +/// - **A topic cycle across the wire is not bounded at all.** A +/// `Locality::Remote` publisher runs both routes, so a nested delivery emits +/// a wire message per nesting level; a peer that echoes amplifies again. That +/// is a property of publish/subscribe — any ROS 2 node that publishes to a +/// topic it subscribes to does the same, and no client library prevents it. +/// Suppressing the wire half for nested deliveries was considered and +/// rejected: a nested publish is a distinct message the callback chose to +/// send, so dropping it would trade a loud problem for a silent one. +/// +/// The guard exists to turn a stack overflow into a dropped message and a +/// greppable log. It is not a cycle detector. +pub const MAX_DELIVERY_DEPTH: u32 = 8; + +thread_local! { + /// Delivery depth for the current thread. Not per channel: a cycle across + /// two topics recurses just as fatally as one topic into itself. + static DELIVERY_DEPTH: Cell = const { Cell::new(0) }; +} + +/// Restores the delivery depth even if a callback panics. +struct DepthGuard; + +impl Drop for DepthGuard { + fn drop(&mut self) { + DELIVERY_DEPTH.with(|d| d.set(d.get().saturating_sub(1))); + } +} + +/// An erased payload. Always an `Arc` for the `T` named by `type_id`. +pub type ErasedPayload = Arc; + +/// Takes the payload **by value**, so the single-subscriber case can move it. +/// +/// With a `&ErasedPayload` the callback had to clone before downcasting, which +/// is a refcount pair on every message. One subscriber is the overwhelmingly +/// common case, and it can be handed the only reference instead. +type LocalCallback = Arc; + +/// A callback that takes the message **owned and mutable**. +/// +/// The shared path hands every receiver the same read-only `Arc`, which is +/// right when several of them want it and wrong when exactly one does: a sole +/// receiver could have been given the value itself, free to mutate or consume +/// it. This is that path. +type OwnedCallback = Arc) + Send + Sync>; + +#[derive(Clone)] +enum Sink { + /// Receives `Arc`; any number may coexist on a topic. + Shared(LocalCallback), + /// Receives `T` by value. Served only when it is the sole subscriber. + Owned(OwnedCallback), +} + +#[derive(Clone)] +struct Entry { + id: u64, + type_id: TypeId, + sink: Sink, +} + +impl Entry { + #[inline] + fn is_shared(&self) -> bool { + matches!(self.sink, Sink::Shared(_)) + } + + #[inline] + fn call_shared(&self, payload: ErasedPayload) { + if let Sink::Shared(cb) = &self.sink { + cb(payload); + } + } +} + +/// The subscriber list for one `(session, topic)`, resolved once. +/// +/// A publisher takes an `Arc` when it is built and never consults the +/// registry again. That matters: looking a topic up per message means hashing a +/// fully-qualified ROS key expression — a long string — on every publish, which +/// at 64 B payloads is a visible share of the whole path. +pub struct Channel { + /// Read on every publish, written only when a subscriber comes or goes. + /// + /// This was an `RwLock>`, and the publish path paid for it twice: + /// the lock itself, and an `Arc` clone of each matching callback so the + /// guard could be dropped before any callback ran. That clone was not + /// optional — invoking a callback under the guard is the re-entrancy + /// deadlock this workspace has fixed repeatedly, because a subscriber + /// callback that publishes is the normal case rather than the exotic one. + /// + /// A snapshot removes both. A publisher loads the current list and calls + /// straight through it; a subscriber coming or going swaps in a new list and + /// leaves any publish already in flight running against the old one. That is + /// the same visibility the clone gave, without the atomics, and it cannot + /// deadlock because nothing is held. + entries: ArcSwap>, +} + +/// What one intra-process publish did. +/// +/// `NoTaker` and `DepthExceeded` both mean nothing was delivered, and they must +/// not be conflated: a caller may fall back to the wire on `NoTaker`, but doing +/// so on `DepthExceeded` re-enters the same callback on a zenoh thread with a +/// fresh depth counter and loops forever. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Delivery { + /// Handed to this many subscribers. + Sent(usize), + /// No subscriber of this type is on the bus. + NoTaker, + /// Refused: delivery is already nested `MAX_DELIVERY_DEPTH` deep. + DepthExceeded, +} + + +/// Which routes one publish took, and what the bus did on its route. +/// +/// A caller cannot reconstruct this from a delivered count: zero means "no +/// taker", "refused at depth" and "there is no bus on this publisher", and the +/// first two must stay distinct — see [`Delivery`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum Published { + /// zenoh only. The audience is not knowable from here. + Wire, + /// The intra-process bus only, with this outcome. + Bus(Delivery), + /// Both, to disjoint audiences: the bus for this session, the wire for + /// everyone else. Only a `Locality::Remote` publisher does this. + BusAndWire(Delivery), +} + +/// Invoke one subscriber callback, isolated. +/// +/// Two things happen here that must happen at every user-code call site. +/// +/// The crate contract in [`crate::reentrancy`] says every invocation of user +/// code is routed through `invoke_user_callback!`, so that calling out while a +/// tracked lock is held is caught in debug builds. Bus delivery is synchronous +/// on the publishing thread, which makes that hazard *more* reachable than the +/// wire path, not less. +/// +/// And a panic is contained — **under `panic = "unwind"`, which is the +/// default**. A build that sets `panic = "abort"` cannot catch anything: the +/// process ends at the panic and this isolation is inert. That is worth knowing +/// before relying on it in an embedded or abort-configured deployment. +/// +/// A panic is contained. On the wire, a panicking callback kills one zenoh +/// task. Here it would unwind into the application's publishing thread and skip +/// every subscriber after it in the snapshot. Delivery order is snapshot order, +/// so which siblings got censored would vary run to run. +fn invoke_isolated(site: &'static str, f: impl FnOnce()) -> bool { + crate::reentrancy::assert_no_guards_held(site); + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(f)) { + Ok(()) => true, + Err(_payload) => { + // Re-raise the crate's own re-entrancy violation instead of + // logging it. `assert_no_guards_held` reports by panicking, and a + // *nested* delivery runs inside this catch_unwind — so without + // this, the detector built to catch "a user callback ran while a + // lock was held" is silently downgraded to a log line in exactly + // the case the bus makes most reachable: a callback that publishes. + #[cfg(debug_assertions)] + // A type, not a message prefix: a subscriber can panic with any + // string it likes, so matching text would let user code force a + // re-raise and defeat the isolation this function provides. + if _payload.downcast_ref::().is_some() { + std::panic::resume_unwind(_payload); + } + tracing::error!( + "[BUS] a subscriber callback panicked during intra-process delivery at {site}; \ + the panic was contained and delivery continued to the remaining subscribers" + ); + false + } + } +} + +impl Channel { + /// Deliver `payload` to every subscriber here whose type matches, returning + /// how many were called. + /// + /// Callbacks are **never** invoked with the lock held. A subscriber callback + /// commonly publishes — that is exactly what the pong side of a ping/pong + /// does — and re-entering under its own read guard is the deadlock this + /// workspace has already fixed three times elsewhere. + /// How many subscribers of `T` on this channel want the message **by value**. + /// + /// A publisher needs this to tell apart two outcomes that both surface as + /// [`Delivery::NoTaker`]: nobody is listening, and somebody is listening + /// whom the shared path structurally cannot serve. [`Channel::publish`] + /// filters on `is_shared`, so an owned subscriber is invisible to it — and + /// an `intra_process_only` publisher has no wire behind the bus to catch the + /// message. Without this, that difference is a silent drop reported as success. + /// + /// It cannot be repaired by handing them a clone: `ZMessage` is + /// `Send + Sync + Sized`, not `Clone`. + pub fn owned_receivers(&self) -> usize + where + T: Any + 'static, + { + let wanted = TypeId::of::(); + self.entries + .load() + .iter() + .filter(|e| e.type_id == wanted && !e.is_shared()) + .count() + } + + pub fn publish(&self, payload: Arc) -> Delivery + where + T: Any + Send + Sync + 'static, + { + // Refuse to recurse without bound. A callback that publishes back onto + // its own topic would otherwise overflow the stack; returning here turns + // that into a dropped message and a loud log, which is recoverable and + // greppable. + let depth = DELIVERY_DEPTH.with(|d| d.get()); + if depth >= MAX_DELIVERY_DEPTH { + tracing::error!( + depth, + max = MAX_DELIVERY_DEPTH, + "intra-process delivery nested too deeply; refusing to recurse further. \ + A subscriber callback is publishing onto a topic that reaches itself. \ + The message was dropped." + ); + return Delivery::DepthExceeded; + } + DELIVERY_DEPTH.with(|d| d.set(depth + 1)); + let _depth_guard = DepthGuard; + + let wanted = TypeId::of::(); + + let entries = self.entries.load(); + let mut matching = entries + .iter() + .filter(|e| e.type_id == wanted) + .filter(|e| e.is_shared()); + let Some(first) = matching.next() else { + return Delivery::NoTaker; + }; + let second = matching.next(); + + let erased: ErasedPayload = payload; + match second { + // One subscriber is the overwhelmingly common case: call straight + // through the snapshot, hand over the only reference, and touch no + // refcount at all beyond the one the caller already holds. + None => { + if invoke_isolated("local_bus::publish", || first.call_shared(erased)) { + Delivery::Sent(1) + } else { + // The only subscriber panicked. Reporting Sent(1) here told + // the caller a message was delivered when none was. + Delivery::NoTaker + } + } + // More than one receiver genuinely needs a reference each, so the + // clones start here and not before. + Some(second) => { + let e1 = erased.clone(); + let e2 = erased.clone(); + let mut count = 0; + count += invoke_isolated("local_bus::publish", || first.call_shared(e1)) as usize; + count += invoke_isolated("local_bus::publish", || second.call_shared(e2)) as usize; + for entry in matching { + let ec = erased.clone(); + count += invoke_isolated("local_bus::publish", || entry.call_shared(ec)) as usize; + } + // The count is subscribers that returned normally, not + // subscribers invoked. A panicking one is logged above. + // + // Zero is NoTaker, never Sent(0): the sole-subscriber arm above + // already reports NoTaker for the same outcome, and two spellings + // of "nothing was delivered" mean a caller that branches on + // NoTaker silently misses one of them. + if count == 0 { + Delivery::NoTaker + } else { + Delivery::Sent(count) + } + } + } + } + + /// Hand `payload` to a sole owning subscriber, by value. + /// + /// Returns the payload back as `Err` when it cannot be delivered that way, + /// so the caller still owns it and can fall back rather than lose it. That + /// happens when no owning subscriber of this type is registered, or when + /// anything else is subscribed as well: with a second receiver the message + /// cannot be given away, and silently downgrading to a shared delivery + /// would defeat the point of having asked for ownership. + pub fn publish_owned(&self, payload: T) -> core::result::Result + where + T: Any + Send + 'static, + { + let wanted = TypeId::of::(); + let entries = self.entries.load(); + + let mut of_type = entries.iter().filter(|e| e.type_id == wanted); + let Some(only) = of_type.next() else { + return Err(payload); + }; + if of_type.next().is_some() { + return Err(payload); + } + let Sink::Owned(cb) = &only.sink else { + return Err(payload); + }; + + let depth = DELIVERY_DEPTH.with(|d| d.get()); + if depth >= MAX_DELIVERY_DEPTH { + tracing::error!( + depth, + max = MAX_DELIVERY_DEPTH, + "intra-process delivery nested too deeply; refusing to recurse further" + ); + // Deliberately NOT Err(payload): the caller treats that as "no + // owning receiver" and falls back, which re-enters this callback on + // a zenoh thread with a fresh depth counter and loops forever. The + // message is dropped, and `DepthExceeded` says so — reporting it as + // a delivery would hide a dropped message behind a success. + return Ok(Delivery::DepthExceeded); + } + DELIVERY_DEPTH.with(|d| d.set(depth + 1)); + let _depth_guard = DepthGuard; + + // Mirror the shared path: a callback that panicked delivered nothing, so + // reporting Sent(1) would tell the caller a message landed when none did. + if invoke_isolated("local_bus::publish_owned", || cb(Box::new(payload))) { + Ok(Delivery::Sent(1)) + } else { + Ok(Delivery::NoTaker) + } + } + + /// How many subscribers this channel has, regardless of type. Diagnostics + /// only; the publish path does not use it. + pub fn subscriber_count(&self) -> usize { + self.entries.load().len() + } +} + +/// Registry of channels, keyed by session then topic. +/// +/// Channels are created on demand and never removed. One empty `Channel` per +/// `(session, topic)` ever used is a bounded, trivial cost, and keeping them +/// means a publisher's handle stays valid across a subscriber coming and going. +type Registry = HashMap>>; + +static BUS: LazyLock> = LazyLock::new(|| RwLock::new(HashMap::new())); +static NEXT_ID: AtomicU64 = AtomicU64::new(0); + +/// Resolve the channel for `(zid, topic)`, creating it if needed. +/// +/// Call this once, when a publisher or subscriber is built — not per message. +/// How many channels the registry currently holds, across every session. +/// +/// Exposed so the reclamation in [`channel`] can be observed. Without a way to +/// read this, "the registry no longer grows without bound" is a claim no test +/// can make: the leak is invisible from the outside, which is why it survived +/// this long. +pub fn total_channels() -> usize { + let bus = match BUS.read() { + Ok(b) => b, + Err(e) => e.into_inner(), + }; + bus.values().map(|topics| topics.len()).sum() +} + +pub fn channel(zid: ZenohId, topic: &str) -> Arc { + // Fast path: it usually exists, and a read lock lets concurrent builders through. + { + let bus = match BUS.read() { + Ok(b) => b, + Err(e) => e.into_inner(), + }; + if let Some(existing) = bus.get(&zid).and_then(|topics| topics.get(topic)) { + return existing.clone(); + } + } + + let mut bus = match BUS.write() { + Ok(b) => b, + Err(e) => e.into_inner(), + }; + + // Reclaim while we already hold the write lock. A channel is removable only + // when the registry is its sole owner and it has no subscribers: every + // `ZPub` and `ZSub` holds its own `Arc`, so a strong count of one + // proves no endpoint can still reach it, and a later builder gets a fresh + // channel that nobody is split from. + // + // Without this the map only ever grows. It is not merely one entry per + // `ZContext` — `channel()` runs in *every* `ZPubBuilder::build()`, so a + // process that publishes on dynamically-named topics accumulates an entry + // per topic, for its lifetime, even after every publisher is dropped. + // Sweep every session, not just this one. Scoping it to `zid` left each + // retired session holding its final channel for the process lifetime, which + // is the session-churn half of the leak rather than a fix for it. + bus.retain(|_, topics| { + topics.retain(|_, ch| Arc::strong_count(ch) > 1 || !ch.entries.load().is_empty()); + !topics.is_empty() + }); + + bus.entry(zid) + .or_default() + .entry(topic.to_owned()) + .or_insert_with(|| { + Arc::new(Channel { + entries: ArcSwap::from_pointee(Vec::new()), + }) + }) + .clone() +} + +/// Keeps a local subscription alive. Dropping it unregisters. +/// +/// `ZSub` holds one, so the registration follows the subscriber's lifetime and a +/// dropped subscriber receives no *further* deliveries. +/// It does not quiesce: a delivery already in flight on another thread runs to +/// completion against the snapshot it loaded, so the callback can still run after +/// `drop` has returned. That is memory-safe, because the snapshot owns the closure +/// and everything it captured. It is not a barrier, so do not tear down a resource +/// the callback merely observes on the strength of having dropped the subscriber. +pub struct LocalSubscription { + channel: Arc, + id: u64, +} + +impl Drop for LocalSubscription { + fn drop(&mut self) { + // Swap in a list without this entry. A publish already in flight keeps + // running against the snapshot it loaded, exactly as it did when the + // list was cloned out from under a lock. + let id = self.id; + self.channel.entries.rcu(|current| { + current + .iter() + .filter(|e| e.id != id) + .cloned() + .collect::>() + }); + } +} + +/// Register a subscriber on `channel` for payloads of type `T`. +pub fn subscribe(channel: Arc, callback: F) -> LocalSubscription +where + T: Any + Send + Sync + 'static, + F: Fn(Arc) + Send + Sync + 'static, +{ + let id = NEXT_ID.fetch_add(1, Ordering::Relaxed); + let erased: LocalCallback = Arc::new(move |payload: ErasedPayload| { + // The publisher already matched on TypeId, so this downcast holds. It is + // still checked rather than assumed — an unchecked cast here would turn + // a bookkeeping bug into undefined behaviour. + match payload.downcast::() { + Ok(typed) => callback(typed), + Err(_) => tracing::error!( + "[LOCAL] payload type did not match subscriber type after a TypeId match" + ), + } + }); + + // Copy on write: rare, and it keeps the publish path free of locks. `rcu` + // may run this closure more than once under contention, so it clones rather + // than moving what it needs. + let type_id = TypeId::of::(); + channel.entries.rcu(|current| { + let mut next = Vec::with_capacity(current.len() + 1); + next.extend(current.iter().cloned()); + next.push(Entry { + id, + type_id, + sink: Sink::Shared(erased.clone()), + }); + next + }); + + debug!("[LOCAL] subscribed id={id}"); + LocalSubscription { channel, id } +} + +/// How many local subscribers `topic` has on `zid`. Diagnostics only. +pub fn subscriber_count(zid: ZenohId, topic: &str) -> usize { + let bus = match BUS.read() { + Ok(b) => b, + Err(e) => e.into_inner(), + }; + bus.get(&zid) + .and_then(|topics| topics.get(topic)) + .map(|c| c.subscriber_count()) + .unwrap_or(0) +} + +/// Register `callback` to receive messages of type `T` **by value**. +/// +/// It is served only when it is the sole subscriber on the channel for that +/// type; see [`Channel::publish_owned`]. Registering one alongside a shared +/// subscriber is allowed, and simply means the owned path cannot give anything +/// away, so the publisher falls back to the shared or wire path. +pub fn subscribe_owned(channel: Arc, callback: F) -> LocalSubscription +where + T: Any + Send + 'static, + F: Fn(T) + Send + Sync + 'static, +{ + let id = NEXT_ID.fetch_add(1, Ordering::Relaxed); + let erased: OwnedCallback = Arc::new(move |payload: Box| { + // The publisher matched on TypeId, so this downcast holds. It is still + // checked: an unchecked cast would turn a bookkeeping bug into + // undefined behaviour. + match payload.downcast::() { + Ok(typed) => callback(*typed), + Err(_) => tracing::error!( + "[LOCAL] payload type did not match an owned subscriber after a TypeId match" + ), + } + }); + + let type_id = TypeId::of::(); + channel.entries.rcu(|current| { + let mut next = Vec::with_capacity(current.len() + 1); + next.extend(current.iter().cloned()); + next.push(Entry { + id, + type_id, + sink: Sink::Owned(erased.clone()), + }); + next + }); + + debug!("[LOCAL] subscribed id={id} (owned)"); + LocalSubscription { channel, id } +} diff --git a/crates/hiroz/src/node.rs b/crates/hiroz/src/node.rs index 86fe8f5e3..4f606600f 100644 --- a/crates/hiroz/src/node.rs +++ b/crates/hiroz/src/node.rs @@ -363,6 +363,8 @@ impl ZNode { keyexpr_format: self.keyexpr_format.clone(), dyn_schema: None, encoding: None, + locality: None, + intra_process_only: false, _phantom_data: Default::default(), } } diff --git a/crates/hiroz/src/parameter/service.rs b/crates/hiroz/src/parameter/service.rs index 23cf67edf..587f27b91 100644 --- a/crates/hiroz/src/parameter/service.rs +++ b/crates/hiroz/src/parameter/service.rs @@ -358,6 +358,8 @@ impl ParameterService { keyexpr_format: keyexpr_format.clone(), dyn_schema: None, encoding: None, + locality: None, + intra_process_only: false, _phantom_data: Default::default(), }; diff --git a/crates/hiroz/src/prelude.rs b/crates/hiroz/src/prelude.rs index c66e74783..3b251a1bc 100644 --- a/crates/hiroz/src/prelude.rs +++ b/crates/hiroz/src/prelude.rs @@ -23,6 +23,7 @@ pub use crate::Builder; /// Core runtime types. pub use crate::context::{ZContext, ZContextBuilder}; pub use crate::node::ZNode; +pub use crate::local_bus::{Delivery, Published}; pub use crate::pubsub::{ZPub, ZSub}; pub use crate::service::{RequestId, ServiceReply, ServiceRequest, ZClient, ZServer}; @@ -55,6 +56,8 @@ pub use crate::parameter::{ /// The `Result` alias used throughout hiroz (equivalent to `zenoh::Result`). pub use zenoh::Result; +/// Locality restriction for publishers and subscribers, re-exported from zenoh. +pub use zenoh::sample::Locality; /// Lifecycle node support. pub use crate::lifecycle::{ diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 5b637d031..3b04dedcf 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -12,6 +12,7 @@ use crate::common::DataHandler; use crate::entity::{EndpointEntity, EndpointKind}; use crate::event::EventsManager; use crate::graph::Graph; +use crate::local_bus::{Delivery, Published}; use crate::impl_with_type_info; use crate::queue::BoundedQueue; use crate::topic_name; @@ -101,6 +102,18 @@ pub(crate) fn apply_transient_local_sub<'a, 'b, 'c, H>( builder } +/// Which routes a publisher's messages take, resolved from what the caller has +/// asserted about the audience. See [`ZPub::route`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Route { + /// The bus alone. The publisher has declared it has no wire. + BusOnly, + /// Both, to disjoint audiences: the bus for this session, the wire for the rest. + BusAndWire, + /// The wire alone. The bus is not consulted. + WireOnly, +} + /// A typed ROS 2-style publisher. Send messages with [`publish`](ZPub::publish) /// (synchronous) or [`async_publish`](ZPub::async_publish) (async). /// @@ -123,6 +136,20 @@ pub struct ZPub { /// If set, this encoding will be used for all published messages. encoding: Option>, graph: Arc, + /// Intra-process bus channel for this topic, resolved once at build time. + /// See [`crate::local_bus`]. + local_channel: Arc, + /// When set, `publish_shared` delivers only to same-session subscribers and + /// never touches zenoh. The caller's assertion that this publisher has no + /// off-session audience — not inferred, because the graph cannot see a + /// plain zenoh subscriber. + intra_process_only: bool, + /// The locality restriction applied to this publisher's wire half, if any. + /// + /// Routing consults it: a `Locality::Remote` wire half cannot reach this + /// session, so the bus and the wire address disjoint audiences and both are + /// used. Without it they overlap, and using both would deliver twice. + locality: Option, _phantom_data: PhantomData<(T, S)>, } @@ -149,6 +176,12 @@ pub struct ZPubBuilder> { /// Encoding format for this publisher. /// If set, all published messages will use this encoding. pub(crate) encoding: Option, + /// Locality restriction for published samples. + /// `None` leaves zenoh's default (`Locality::Any`) in place. + pub(crate) locality: Option, + /// Restrict `publish_shared` to the intra-process bus. See + /// [`ZPubBuilder::with_intra_process_only`]. + pub(crate) intra_process_only: bool, pub(crate) _phantom_data: PhantomData<(T, S)>, } @@ -217,6 +250,94 @@ impl ZPubBuilder { self } + /// Set the locality restriction for this publisher. + /// + /// This restricts which matching subscribers receive the published samples: + /// those in the same session, those in other sessions, or both (the + /// default). + /// + /// [`Locality::SessionLocal`](zenoh::sample::Locality::SessionLocal) is the intra-process fast path. Zenoh skips + /// the transport entirely for it — no link, no wire encode, no shared + /// memory — and hands the payload straight to the matching subscriber + /// callbacks in this session. Every node created from one [`ZContext`] + /// shares one zenoh session, so this reaches sibling nodes in the same + /// process. + /// + /// [`ZContext`]: crate::context::ZContext + /// + /// # This makes the publisher invisible off-process + /// + /// A [`Locality::SessionLocal`](zenoh::sample::Locality::SessionLocal) publisher is not reachable from any other + /// process, including `ros2 topic echo`. Set it only where the subscriber + /// is known to live in the same context. + /// + /// # Serialization still happens + /// + /// Zenoh payloads are bytes, so the message is still CDR-encoded on publish + /// and decoded on receive. This setting removes the transport, not the + /// serialization. Publisher SHM is wasted work under it — pair it with + /// [`without_shm`](Self::without_shm). + /// + /// # Example + /// + /// ```no_run + /// use zenoh::sample::Locality; + /// use hiroz::Builder; + /// + /// # fn main() -> zenoh::Result<()> { + /// # let ctx = hiroz::context::ZContextBuilder::default().build()?; + /// # let node = ctx.create_node("test").build()?; + /// let publisher = node + /// .create_pub::("topic") + /// .with_locality(Locality::SessionLocal) + /// .without_shm() + /// .build()?; + /// # Ok(()) + /// # } + /// ``` + pub fn with_locality(mut self, locality: zenoh::sample::Locality) -> Self { + self.locality = Some(locality); + self + } + + /// **Prototype.** Restrict [`ZPub::publish_shared`] to the intra-process + /// bus: the message is handed to same-session subscribers as an `Arc` + /// and never goes near zenoh, so it is neither serialized nor transmitted. + /// + /// This affects `publish_shared` only. [`ZPub::publish`] keeps its normal + /// wire behaviour whatever this is set to. + /// + /// # Why the flag exists, and why it should not + /// + /// The caller asserts the audience; it is not inferred. Reading it off the +/// ROS graph was tried and withdrawn: a plain zenoh subscriber declares no +/// liveliness token, so the graph cannot see it, and a publisher that +/// concluded "everyone is local" from the graph dropped that subscriber's +/// messages with nothing reporting the loss. + /// every other process, including `ros2 topic echo`, and to any local + /// subscriber that did not register on the bus with + /// [`ZSubBuilder::build_with_shared_callback`]. Set it only where both ends + /// are known. + /// # This publisher still appears in the ROS graph + /// + /// The liveliness token is declared in `build()` before any locality is + /// consulted, so a publisher that can reach nothing outside this process + /// still advertises itself as an ordinary ROS 2 endpoint. Every other node + /// counts it: `ros2 topic info -v`, `ros2 node info`, a remote subscriber's + /// matched-publisher event, and any `Graph::count` derived from them. + /// `ros2 topic echo` shows the topic and never a message. + /// + /// Suppressing the token would make the counts honest and remove the + /// publisher from `ros2 node info`, which is a change to graph visibility + /// and belongs in its own change with its own test. Until then this is a + /// disclosed limitation, not an oversight: a node debugging "the publisher + /// is there but nothing arrives" has no signal separating this from a QoS + /// mismatch. + pub fn with_intra_process_only(mut self) -> Self { + self.intra_process_only = true; + self + } + pub fn with_serdes(self) -> ZPubBuilder { ZPubBuilder { entity: self.entity, @@ -228,6 +349,8 @@ impl ZPubBuilder { keyexpr_format: self.keyexpr_format.clone(), dyn_schema: self.dyn_schema, encoding: self.encoding, + locality: self.locality, + intra_process_only: self.intra_process_only, _phantom_data: PhantomData, } } @@ -334,6 +457,14 @@ where } } + // Apply the locality restriction before .advanced(); AdvancedPublisher + // forwards its own destination onto this inner builder, so setting it + // here is what survives. + if let Some(locality) = self.locality { + pub_builder = pub_builder.allowed_destination(locality); + debug!("[PUB] Locality restriction: {:?}", locality); + } + // Build an AdvancedPublisher and apply TransientLocal config if needed. let pub_builder = pub_builder.advanced(); debug!( @@ -362,6 +493,8 @@ where debug!("[PUB] Using encoding: {}", enc); } + let local_channel = crate::local_bus::channel(self.session.zid(), &qualified_topic); + Ok(ZPub { entity: self.entity, sn: AtomicUsize::new(0), @@ -375,6 +508,9 @@ where dyn_schema: self.dyn_schema, encoding, graph: self.graph, + local_channel, + intra_process_only: self.intra_process_only, + locality: self.locality, _phantom_data: Default::default(), }) } @@ -559,6 +695,224 @@ where put_builder.await } + /// **Prototype.** Publish an already-shared message without serializing it. + /// + /// Every subscriber in the same zenoh session that registered through + /// [`ZSubBuilder::build_with_shared_callback`] for this exact `T` receives a + /// clone of the `Arc` — a refcount bump. The payload is not encoded and its + /// bytes are not copied. + /// + /// Reports which routes ran, as a [`Published`]. + /// + /// # Which path a message takes + /// + /// The route follows what the **caller has asserted about the audience**. + /// It is never inferred from the graph. + /// + /// | the publisher says | route | returns | + /// |---|---|---| + /// | `with_intra_process_only()` | the bus alone; nothing goes on the wire | `Bus(Sent(n))`, or `Bus(NoTaker)` when nobody is listening | + /// | `with_locality(Remote)` | both, to disjoint audiences | `BusAndWire(..)` | + /// | neither | the wire alone; the bus is not consulted | `Wire` | + /// + /// Row three is the default and is deliberately conservative. A subscriber + /// that the ROS graph cannot see — a plain `z_sub`, a storage plugin, a + /// recorder — is reachable only over the wire, so a publisher that has not + /// been told the audience takes it. + /// + /// `Bus(NoTaker)` is a distinct outcome rather than a silent drop: the + /// caller asserted there was a local audience and there was none, which is + /// worth acting on. See [`crate::local_bus`]. + /// + /// # Ordering against [`publish`](Self::publish) + /// + /// Bus delivery is synchronous on the calling thread: the subscriber + /// callbacks have run by the time this returns. A message sent with + /// `publish` travels the wire and arrives later. Interleaving the two on one + /// topic gives no ordering guarantee between them. + /// + /// # QoS is not carried on the intra-process path + /// + /// A message that goes to the bus does not enter the wire publisher's + /// cache and carries no attachment. Concretely: + /// + /// - **`TRANSIENT_LOCAL` is refused, not silently violated.** Nothing on + /// the bus enters the durability cache, so a late joiner would be served + /// the surviving samples *as if they were a complete history*. A + /// transient-local publisher that has no wire is rejected instead. + /// - `RELIABLE` / `BEST_EFFORT` and `KEEP_LAST(n)` have no meaning here: + /// delivery is a synchronous inline call with no queue, so nothing is + /// ever dropped for congestion and no depth is ever displaced. + /// - The attachment — source gid, sequence number, source timestamp — is + /// absent, so no bus message carries ROS message-info. + /// + pub fn publish_shared(&self, msg: Arc) -> Result + where + T: Send + Sync + 'static, + { + match self.route() { + Route::BusOnly => { + // TRANSIENT_LOCAL lives in the wire publisher's cache and this + // publisher has no wire. Serving the bus would hand a late + // joiner whatever predates this call *as if it were the + // promised history* — wrong, rather than absent. Refuse, and + // let the caller choose. See #133. + if let Some(e) = self.refuse_durable_bus() { + return Err(e); + } + let d = self.local_channel.publish(msg); + if d == Delivery::NoTaker { + // Distinguish "nobody is listening" from "somebody is + // listening whom this path cannot serve". The shared path + // filters on is_shared(), so an owned subscriber is skipped + // — and this publisher has no wire to catch the message. + // Reporting Ok here loses it silently. + let owned = self.local_channel.owned_receivers::(); + if owned > 0 { + return Err(format!( + "publish_shared on an intra-process-only publisher for {}: \ + {owned} subscriber(s) on this topic take the message by value, \ + and a shared publish cannot serve them. The message would be \ + dropped with no wire behind it. Use publish_owned(), or build \ + the subscriber with build_with_shared_callback().", + self.entity.topic + ) + .into()); + } + debug!( + "[PUB] intra-process only, no local subscriber on {}: message dropped", + self.entity.topic + ); + } + // DepthExceeded is kept distinct from "nothing delivered": a + // caller may fall back to the wire on NoTaker and must not on + // DepthExceeded. + Ok(Published::Bus(d)) + } + // Both audiences, and no subscriber is on both. TRANSIENT_LOCAL is + // safe here: the wire publish populates the cache as it always did. + Route::BusAndWire => { + // Wire first, matching publish_owned. Bus delivery is + // synchronous, so running it first and then failing on the wire + // returns Err *after* every local subscriber has already been + // called — and Result has no partial-success value + // to say so. A caller that retries then delivers twice locally. + self.publish(&msg)?; + Ok(Published::BusAndWire(self.local_channel.publish(msg))) + } + Route::WireOnly => { + self.publish(&msg)?; + Ok(Published::Wire) + } + } + } + + /// Which routes this publisher's messages take. + /// + /// Resolved in one place because it is asserted by the caller and read by + /// every publishing method. When each method decided for itself they drifted: + /// `publish_owned` on a `Locality::Remote` publisher took the bus and + /// returned, losing the message for every off-session subscriber, while + /// `publish_shared` in the identical configuration served both. + /// + /// `with_intra_process_only()` wins over a locality, because it is the + /// stronger assertion: it says this publisher has no wire at all. + fn route(&self) -> Route { + if self.intra_process_only { + Route::BusOnly + } else if self.locality == Some(zenoh::sample::Locality::Remote) { + // The wire half cannot reach this session, so the two halves address + // disjoint audiences and both must run. + Route::BusAndWire + } else { + Route::WireOnly + } + } + + /// Refuse a durable publisher the bus, which has no durability cache. + /// + /// This lives in one place because it must guard **every** route onto the + /// bus. It was originally written inline in `publish_shared`, and + /// `publish_owned` reached the bus around it — the same violation, through + /// the sibling method. See #133. + fn refuse_durable_bus(&self) -> Option { + if !self.intra_process_only { + // A `Locality::Remote` publisher still puts every message on the + // wire, so its durability cache is populated as it always was. + return None; + } + if !matches!(self.entity.qos.durability, QosDurability::TransientLocal) { + return None; + } + Some( + format!( + "publish on a TRANSIENT_LOCAL, intra-process-only publisher for {}: \ + the intra-process path has no durability cache, so a late-joining \ + subscriber would be served a history this message is missing from. \ + Use publish(), or drop the transient-local durability.", + self.entity.topic + ) + .into(), + ) + } + + /// Publish `msg` by value, giving it away when exactly one receiver wants it. + /// + /// Where [`publish_shared`](Self::publish_shared) hands every receiver the + /// same read-only `Arc`, this hands a sole owning subscriber the value + /// itself — free to mutate it, consume it, or put it back in a pool. It is + /// the move rclcpp performs when a topic has exactly one taker. + /// + /// The route is the same one [`publish_shared`](Self::publish_shared) + /// documents, resolved from what the caller asserted. Within a bus route, + /// a sole owning subscriber is given the value; with more than one taker + /// there is nothing to give away, so it is shared instead. + /// + /// Returns a [`Published`] saying which routes ran — not a count. An + /// `intra_process_only` publisher with no listener reports + /// `Bus(Delivery::NoTaker)` and the message is dropped; it does **not** + /// fall through to the wire, because such a publisher has none. + /// + pub fn publish_owned(&self, msg: T) -> Result + where + T: Send + Sync + 'static, + { + match self.route() { + // The wire needs the value in order to serialize it and the bus + // wants to own it, so the wire goes first from a reference and the + // value is given away afterwards. Sharing instead would reach the + // wire but stop serving owned subscribers, because the shared path + // filters on `is_shared()`. + Route::BusAndWire => { + self.publish(&msg)?; + Ok(Published::BusAndWire( + match self.local_channel.publish_owned(msg) { + Ok(d) => d, + // No sole owning receiver; the shared half may want it. + Err(returned) => self.local_channel.publish(Arc::new(returned)), + }, + )) + } + Route::BusOnly => { + if let Some(e) = self.refuse_durable_bus() { + return Err(e); + } + match self.local_channel.publish_owned(msg) { + Ok(d) => Ok(Published::Bus(d)), + // Handed back untouched: not exactly one owning receiver. + // Share it instead — publish_shared refuses rather than + // dropping if the only subscribers are owned ones. + Err(returned) => self.publish_shared(Arc::new(returned)), + } + } + Route::WireOnly => { + self.publish(&msg)?; + Ok(Published::Wire) + } + } + } + + /// Publish pre-serialized data directly /// /// Accepts any type that implements `Into`: @@ -863,10 +1217,117 @@ where graph: self.graph, dyn_schema: self.dyn_schema, expected_encoding: self.expected_encoding, + _local_sub: None, _phantom_data: Default::default(), }) } + /// Build a subscriber whose callback receives the message **by value**. + /// + /// Served only by [`ZPub::publish_owned`], and only when this is the sole + /// subscriber on the topic for this type — with a second receiver the + /// message cannot be given away. Otherwise the publisher falls back to a + /// shared or wire delivery, which this subscriber does not receive. + /// + /// Use it where the receiver wants to mutate or consume the message, which + /// the shared path cannot offer. + pub fn build_with_owned_callback(self, callback: F) -> Result> + where + T: Send + Sync + 'static, + F: Fn(T) + Send + Sync + 'static, + S: for<'a> ZDeserializer = &'a [u8], Output = T> + 'static, + { + let zid = self.session.zid(); + // Both halves run the same callback. The wire half is not decoration: + // without it this subscriber discards every message that does not + // arrive by `publish_owned` on the bus — including from every remote + // publisher — while still advertising a live subscription. + // + // It cannot deliver twice: a publisher takes the bus or the wire for + // any one message, never both, unless its wire half is `Remote`- + // restricted, in which case the two audiences are disjoint. + let callback = Arc::new(callback); + let wire_half = callback.clone(); + // Read before `self` is consumed below. + let bus_is_excluded = self.locality == Some(zenoh::sample::Locality::Remote); + let mut sub = self.build_with_callback(move |m: T| wire_half(m))?; + // A `Locality::Remote` subscriber has asked not to see same-session + // traffic, and the bus is same-session by definition. Registering it + // here would deliver exactly what its own `allowed_origin` filter + // excludes — the wire half honours the restriction while the bus half + // silently ignored it. Skip the bus registration; the wire half keeps + // serving the remote traffic this subscriber asked for. + if bus_is_excluded { + return Ok(sub); + } + + let topic = sub.entity.topic.clone(); + let channel = crate::local_bus::channel(zid, &topic); + sub._local_sub = Some(crate::local_bus::subscribe_owned::( + channel, + move |m: T| callback(m), + )); + Ok(sub) + } + + /// **Prototype.** Build a subscriber that receives an `Arc` from a + /// same-session publisher without any serialization, and still receives + /// remote traffic over the wire as usual. + /// + /// Two registrations are made: + /// + /// | path | source | cost | + /// |---|---|---| + /// | intra-process bus | a same-session [`ZPub::publish_shared`] for this exact `T` | a refcount bump | + /// | zenoh subscriber, no origin filter | anything on the wire, near or far | the usual CDR decode | + /// + /// The wire half applies no origin filter, and does not need one. A + /// publisher takes the bus or the wire for any one message, never both — + /// unless its wire half is `Locality::Remote`, in which case the two + /// audiences are disjoint and no subscriber sees it twice. Suppressing the + /// duplicate is the publisher's job, because it is the only side that knows + /// which routes it used. + /// + /// The type match is exact: a publisher of a different concrete Rust type + /// on the same topic is not delivered here, even if the ROS type name + /// agrees. See [`crate::local_bus`]. + pub fn build_with_shared_callback(self, callback: F) -> Result> + where + T: Send + Sync + 'static, + F: Fn(Arc) + Send + Sync + 'static, + S: for<'a> ZDeserializer = &'a [u8], Output = T> + 'static, + { + let zid = self.session.zid(); + + let callback = Arc::new(callback); + let wire_cb = callback.clone(); + // Read before `self` is consumed below. + let bus_is_excluded = self.locality == Some(zenoh::sample::Locality::Remote); + let mut sub = self.build_with_callback(move |msg: T| (*wire_cb)(Arc::new(msg)))?; + + // build_with_callback qualified the topic, so read it back rather than + // re-deriving it — the publisher keys the bus on its own qualified topic + // and the two must agree exactly. + // A `Locality::Remote` subscriber has asked not to see same-session + // traffic, and the bus is same-session by definition. Registering it + // here would deliver exactly what its own `allowed_origin` filter + // excludes — the wire half honours the restriction while the bus half + // silently ignored it. Skip the bus registration; the wire half keeps + // serving the remote traffic this subscriber asked for. + if bus_is_excluded { + return Ok(sub); + } + + let topic = sub.entity.topic.clone(); + let channel = crate::local_bus::channel(zid, &topic); + sub._local_sub = Some(crate::local_bus::subscribe::( + channel, + move |arc: Arc| (*callback)(arc), + )); + debug!("[SUB] intra-process bus registered: topic={topic}"); + Ok(sub) + } + /// Build a subscriber with a callback that processes deserialized messages directly. /// /// This method creates a subscriber that invokes the provided callback for each @@ -979,6 +1440,15 @@ pub struct ZSub { pub dyn_schema: Option>, /// Expected encoding for validation. pub expected_encoding: Option, + /// Registration on the intra-process bus, when built with + /// [`ZSubBuilder::build_with_shared_callback`]. Dropping it unregisters, so + /// a dropped subscriber receives no *further* deliveries. + /// It does not quiesce: a delivery already in flight on another thread runs to + /// completion against the snapshot it loaded, so the callback can still run after + /// `drop` has returned. That is memory-safe, because the snapshot owns the closure + /// and everything it captured. It is not a barrier, so do not tear down a resource + /// the callback merely observes on the strength of having dropped the subscriber. + _local_sub: Option, _phantom_data: PhantomData<(T, Q, S)>, } diff --git a/crates/hiroz/src/reentrancy.rs b/crates/hiroz/src/reentrancy.rs index 0127262ac..d8f78635f 100644 --- a/crates/hiroz/src/reentrancy.rs +++ b/crates/hiroz/src/reentrancy.rs @@ -87,6 +87,32 @@ pub fn live_guards() -> usize { } } +/// The payload of a re-entrancy violation panic. +/// +/// [`local_bus`](crate::local_bus) contains subscriber panics so one bad +/// callback cannot censor its siblings. That containment must not swallow +/// *this* panic: it is the crate reporting its own contract violation, and a +/// nested delivery raises it from inside that `catch_unwind`. +/// +/// It is a type rather than a message prefix because the alternative is +/// forgeable: a subscriber that panicked with a `String` beginning with the +/// prefix would be re-raised as though it were a violation, breaking the very +/// isolation guarantee the containment exists to provide. The private field +/// means no code outside this crate can construct one. +#[derive(Debug)] +#[non_exhaustive] +pub struct ReentrancyViolation { + /// The operator-facing description. + pub message: String, +} + +impl std::fmt::Display for ReentrancyViolation { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } +} + + /// Panics (debug only) if any tracked guard is live on this thread. /// /// Call immediately before invoking user code. `site` is reproduced in the panic @@ -97,14 +123,17 @@ pub fn assert_no_guards_held(site: &str) { #[cfg(debug_assertions)] { let live = live_guards(); - assert!( - live == 0, - "hiroz re-entrancy rule violated at `{site}`: about to invoke a user \ - callback with {live} lock guard(s) live on this thread. A callback \ - that re-enters hiroz will deadlock if it touches a lock this thread \ - holds. Fix: collect what you need into an owned value, drop every \ - guard, then invoke the callback." - ); + if live != 0 { + std::panic::panic_any(ReentrancyViolation { + message: format!( + "hiroz re-entrancy rule violated at `{site}`: about to invoke a user \ + callback with {live} lock guard(s) live on this thread. A callback \ + that re-enters hiroz will deadlock if it touches a lock this thread \ + holds. Fix: collect what you need into an owned value, drop every \ + guard, then invoke the callback." + ), + }); + } } #[cfg(not(debug_assertions))] let _ = site; @@ -267,15 +296,38 @@ mod tests { } /// A tripwire that never fires is indistinguishable from a clean codebase. + /// + /// The payload is asserted by **type**, not by message. `should_panic` can + /// only match a string, and matching the message is exactly what made the + /// old classifier forgeable: a subscriber panicking with the same words + /// would have been mistaken for a violation. Testing the type is both + /// stronger and the thing the containment in `local_bus` actually keys on. #[test] - #[cfg_attr(debug_assertions, should_panic(expected = "re-entrancy rule violated"))] fn assert_fires_while_a_guard_is_live() { - let m = TrackedMutex::new(0u32); - let _g = m.lock().unwrap(); - assert_no_guards_held("deliberate violation"); - // Compiled out in release, so no panic is expected there. + let fired = std::panic::catch_unwind(|| { + let m = TrackedMutex::new(0u32); + let _g = m.lock().unwrap(); + assert_no_guards_held("deliberate violation"); + }); + + #[cfg(debug_assertions)] + { + let payload = fired.expect_err("the tripwire did not fire while a guard was live"); + let violation = payload + .downcast_ref::() + .expect("the panic must carry ReentrancyViolation, not a bare string"); + assert!( + violation.message.contains("deliberate violation"), + "the site must reach the operator: {}", + violation.message + ); + } + // Compiled out in release, so nothing panics there. #[cfg(not(debug_assertions))] - assert_eq!(live_guards(), 0); + { + assert!(fired.is_ok(), "the check is debug-only"); + assert_eq!(live_guards(), 0); + } } /// The underflow assertion needs the same proof the tripwire gets. Only this