diff --git a/crates/hiroz-tests/tests/hu_meter.rs b/crates/hiroz-tests/tests/hu_meter.rs index acfdeed5f..5b561fb24 100644 --- a/crates/hiroz-tests/tests/hu_meter.rs +++ b/crates/hiroz-tests/tests/hu_meter.rs @@ -1009,20 +1009,40 @@ fn test_hu_meter_param_load() { // ─── echo --timeout ────────────────────────────────────────────────────────── +/// `echo` on a topic nobody publishes: the contract is **terminate, and say +/// why**. It used to exit 0 after the timeout having printed nothing, which is +/// the exact ambiguity this branch removes — a silent success is +/// indistinguishable from "the topic is idle". There is no schema to resolve +/// (nothing advertises a type and `subscribe` carries only a topic name), so +/// `subscribe` now fails, hu-meter reports it and exits non-zero. +/// +/// The "does not hang" half of the original assertion is preserved by the fact +/// that `run_hu_meter` returns at all. +/// +/// This is also the end-to-end cover for the epoch budget: with no publisher, +/// `subscribe` blocks for the *full* discovery timeout (5 s) inside a guest +/// dispatch whose epoch deadline is ~3 s of wall clock. Without +/// `HostBlockGuard` suspending the epoch ticker, the guest traps on return, the +/// error branch below never runs, and `exit_code` stays `None` — the command +/// would print nothing and hang in the tick loop instead of failing here. #[test] #[serial_test::serial] -fn test_hu_meter_echo_timeout_exits() { +fn test_hu_meter_echo_no_publisher_reports_and_exits() { let router = TestRouter::new(); - // No publisher — echo should exit after the timeout rather than hang. let out = run_hu_meter( router.endpoint(), &["echo", "/no_publisher_topic", "--timeout", "1"], ); - // Should exit cleanly (not hang indefinitely). + let stderr = String::from_utf8_lossy(&out.stderr); assert!( - out.status.success(), - "hu meter echo --timeout should exit cleanly when no messages arrive: {}", - String::from_utf8_lossy(&out.stderr) + !out.status.success(), + "hu meter echo on a topic with no publisher must report a failure, not exit \ + 0 having printed nothing (stdout: {}, stderr: {stderr})", + String::from_utf8_lossy(&out.stdout) + ); + assert!( + stderr.contains("/no_publisher_topic"), + "the failure must name the topic it could not resolve: {stderr}" ); } diff --git a/crates/hiroz-union/plugins/hu-meter/src/lib.rs b/crates/hiroz-union/plugins/hu-meter/src/lib.rs index 371cd45f2..adcf6737b 100644 --- a/crates/hiroz-union/plugins/hu-meter/src/lib.rs +++ b/crates/hiroz-union/plugins/hu-meter/src/lib.rs @@ -16,21 +16,34 @@ struct HuMeter { // Ticks elapsed (used for duration tracking at tick_ms = 1000 ms) ticks: u32, duration_ticks: u32, + /// Keeps `hz`/`bw` visible in the ROS graph. The rate tracker behind + /// `measure_hz` is a raw zenoh wildcard subscriber with no liveliness + /// token, so it announces nothing -- and a publisher that gates on + /// `wait_for_subscription` then waits forever and never publishes, which + /// reads back as a topic with no traffic. Holding a real subscription + /// restores the announcement. Its messages are unused; dropping it would + /// undeclare the token, so it must live as long as the measurement. + /// + /// Best-effort on purpose: `None` when the topic advertises no type, where + /// `hz` and `bw` must still work. + graph_presence: Option, } enum Mode { /// Waiting for startup event (initial state) Init, - /// Measure publish rate on a topic - Hz { - topic: String, - sub: Option, - }, - /// Measure bandwidth on a topic - Bw { - topic: String, - sub: Option, - }, + /// Measure publish rate on a topic. The numbers come from + /// `ros::measure_hz`, backed by a raw wildcard subscriber in the host that + /// counts and sizes bytes without decoding them -- so `hz` needs no schema + /// and keeps working on a topic whose type cannot be resolved. + /// + /// A best-effort subscription is still taken, in `HuMeter::graph_presence` + /// rather than here, purely so `hz` announces itself in the ROS graph. It + /// is deliberately not part of this variant: the measurement does not + /// depend on it, and a failure to acquire it must not fail the command. + Hz { topic: String }, + /// Measure bandwidth on a topic. See `Hz`; `bw` works the same way. + Bw { topic: String }, /// Echo messages Echo { topic: String, @@ -78,6 +91,7 @@ impl HuMeter { json: false, ticks: 0, duration_ticks: 0, + graph_presence: None, } } @@ -163,19 +177,13 @@ impl HuMeter { return; }; self.duration_ticks = duration_ticks; - let sub = match ros::subscribe(&topic) { - Ok(s) => s, - Err(e) => { - render::eprintln(&format!("Failed to subscribe to {topic}: {e}")); - render::exit(1); - self.mode = Mode::Done; - return; - } - }; - self.mode = Mode::Hz { - topic, - sub: Some(sub), - }; + // The numbers come from `measure_hz`, not from this subscription -- see + // `Mode::Hz`. It exists so `hz` still ANNOUNCES itself in the graph, as + // it did when this took a typed subscription for its data. Errors are + // ignored: on a topic with no advertised type there is nothing to + // announce, and `hz` must keep working there. + self.graph_presence = ros::subscribe(&topic).ok(); + self.mode = Mode::Hz { topic }; } fn cmd_bw(&mut self, args: &[String]) { @@ -187,19 +195,10 @@ impl HuMeter { return; }; self.duration_ticks = duration_ticks; - let sub = match ros::subscribe(&topic) { - Ok(s) => s, - Err(e) => { - render::eprintln(&format!("Failed to subscribe to {topic}: {e}")); - render::exit(1); - self.mode = Mode::Done; - return; - } - }; - self.mode = Mode::Bw { - topic, - sub: Some(sub), - }; + // Graph presence only, exactly as in `cmd_hz`; `measure_bw` supplies + // the numbers. + self.graph_presence = ros::subscribe(&topic).ok(); + self.mode = Mode::Bw { topic }; } fn cmd_echo(&mut self, args: &[String]) { @@ -270,6 +269,13 @@ impl HuMeter { return; } }; + // Say that we are listening. A subscribe that succeeds and then prints + // nothing is indistinguishable from a broken one: the user cannot tell + // "no traffic on this topic" from "this tool is not working". stderr, + // not stdout, so `--json | jq` and friends stay clean. + render::eprintln(&format!( + "subscribed to {topic}; waiting for messages (Ctrl-C to stop)" + )); self.mode = Mode::Echo { topic, sub: Some(sub), @@ -1265,7 +1271,7 @@ impl HuMeter { let done = self.duration_ticks > 0 && self.ticks >= self.duration_ticks; match &mut self.mode { - Mode::Hz { topic, sub } => { + Mode::Hz { topic } => { let window_ms = 1000u32; match ros::measure_hz(topic, window_ms) { Ok(m) => { @@ -1283,13 +1289,12 @@ impl HuMeter { } Err(e) => render::println(&format!("measure-hz error: {e}")), } - let _ = sub; // keep subscription alive if done { render::exit(0); self.mode = Mode::Done; } } - Mode::Bw { topic, sub } => { + Mode::Bw { topic } => { let window_ms = 1000u32; match ros::measure_bw(topic, window_ms) { Ok(m) => { @@ -1307,7 +1312,6 @@ impl HuMeter { } Err(e) => render::println(&format!("measure-bw error: {e}")), } - let _ = sub; if done { render::exit(0); self.mode = Mode::Done; diff --git a/crates/hiroz-union/src/plugin/wasm/host/ros.rs b/crates/hiroz-union/src/plugin/wasm/host/ros.rs index 18a77d36d..04dc425f0 100644 --- a/crates/hiroz-union/src/plugin/wasm/host/ros.rs +++ b/crates/hiroz-union/src/plugin/wasm/host/ros.rs @@ -3,9 +3,15 @@ use std::sync::Arc; use std::time::Duration; -use hiroz::dynamic::{ - DynamicMessage, DynamicValue, FieldType, MessageSchema, - serialization::{deserialize_cdr, serialize_cdr}, +use hiroz::{ + Builder, + dynamic::{ + DynSub, DynamicMessage, DynamicValue, FieldType, MessageSchema, + MessageSchemaTypeDescription, + serialization::{deserialize_cdr, serialize_cdr}, + }, + graph::Graph, + node::ZNode, }; use wasmtime::component::Resource; use zenoh::Wait; @@ -16,6 +22,149 @@ use super::super::state::{PluginState, ServiceClientData, SubscriptionData}; use super::hu; use hu::plugin::types::PluginError; +/// The message type advertised by a live publisher or subscriber on `topic`, if +/// any. Free-standing (rather than a `PluginState` method) so it can be tested +/// against a hand-built `Graph` without a wasmtime store. +fn live_topic_type_info(graph: &Graph, topic: &str) -> Option { + use hiroz_protocol::{EndpointKind, Entity}; + [EndpointKind::Publisher, EndpointKind::Subscription] + .into_iter() + .find_map(|kind| { + graph + .get_entities_by_topic(kind, topic) + .first() + .and_then(|ent| match ent.as_ref() { + Entity::Endpoint(ep) => ep.type_info.clone(), + _ => None, + }) + }) +} + +/// The three outcomes of checking a local `.msg` against a topic's advertised +/// hash. "Not advertised" is deliberately distinct from "mismatch": it cannot be +/// verified either way, so it must not be refused. +#[derive(Debug, PartialEq, Eq)] +enum HashCheck { + NotAdvertised, + Match, + Mismatch, +} + +/// Compare two hashes by value. +/// +/// Extracted so the three-way decision is testable without a router, a graph or +/// a `.msg` on disk. This guard has been wrong twice, both times in the +/// comparison rather than the surrounding plumbing, and both times found by +/// reading rather than by a test. +fn check_type_hash(local: &hiroz::TypeHash, advertised: &hiroz::TypeHash) -> HashCheck { + if *advertised == hiroz::TypeHash::zero() { + HashCheck::NotAdvertised + } else if local == advertised { + HashCheck::Match + } else { + HashCheck::Mismatch + } +} + +/// Render a `hiroz_protocol::TypeHash` as RIHS, without the `no-type-hash` gate. +/// +/// `TypeHash::to_rihs_string` collapses to the constant "TypeHashNotSupported" +/// under that feature. That is right for the wire, where the constant *is* the +/// representation, and wrong for a diagnostic, which must show what the peer +/// actually advertised. +fn rihs_ungated(hash: &hiroz::TypeHash) -> String { + let hex: String = hash.value.iter().map(|b| format!("{b:02x}")).collect(); + format!("RIHS{:02x}_{hex}", hash.version) +} + +/// Build a dynamic subscriber for `topic` from a `.msg` on `HIROZ_MSG_PATH`, +/// used only when live discovery has already failed. The type *name* is still +/// taken from the graph -- `subscribe(topic)` carries no type, so without a live +/// endpoint there is nothing to look up. +/// +/// Returns the specific reason on failure rather than a bare `None`: the three +/// ways this can fail (no live endpoint, no `.msg` on disk, the subscriber +/// declaration itself failing) send the reader to three different places, and +/// collapsing them into one "check `HIROZ_MSG_PATH`" message is the same class +/// of misdirection this whole path exists to remove. +fn dyn_sub_from_local_msg(node: &ZNode, graph: &Graph, topic: &str) -> Result { + let Some(ti) = live_topic_type_info(graph, topic) else { + return Err(format!( + "no publisher or subscriber on {topic} advertises a type, so there is no \ + type name to look up -- `subscribe` carries only a topic, and these \ + commands have no --type flag" + )); + }; + let canonical = hiroz::dynamic::ros_type_name_from_dds(&ti.name); + let Some(schema) = hiroz::dynamic::load_schema(&canonical) else { + return Err(format!( + "no .msg for {canonical} (advertised by {topic}) was found on HIROZ_MSG_PATH" + )); + }; + + // The subscriber's key expression carries the *publisher's* hash, so messages + // arrive even when the local .msg disagrees. CDR is positional, so a skewed + // schema yields plausible, wrong field values rather than a decode error. + // Refuse instead. + // + // Compare VALUES, never rendered strings: hiroz_protocol's renderer is gated + // on `no-type-hash` and then returns one constant for every value, so a + // string comparison passes on anything. Convert through the RIHS01 string -- + // the schema-side renderer and protocol-side parser are both ungated. + // + // "No hash advertised" is a third state, not a mismatch: unverifiable either + // way, so warn and continue. + let local_schema_hash = schema + .compute_type_hash() + .map_err(|e| format!("could not hash the local .msg for {canonical}: {e}"))?; + let local = hiroz::TypeHash::from_rihs_string(&local_schema_hash.to_rihs_string()) + .unwrap_or_else(hiroz::TypeHash::zero); + + match check_type_hash(&local, &ti.hash) { + // A publisher that advertises no hash -- a Humble node, or any peer + // built without type hashing. "Not advertised" is a third state, not a + // mismatch: it cannot be verified either way. Refusing here told the + // user to point HIROZ_MSG_PATH at definitions hashing to zero, which + // no .msg does, and made every cross-distro case fail -- the headline + // case for having a disk fallback at all. + HashCheck::NotAdvertised => tracing::warn!( + "{topic} advertises no type hash, so the local .msg for {canonical} \ + cannot be verified against it; decoding with the local definition" + ), + HashCheck::Mismatch => { + return Err(format!( + "local .msg for {canonical} hashes to {} but {topic} advertises \ + {}; refusing to decode with a mismatched schema -- point \ + HIROZ_MSG_PATH at the message definitions the publisher was built from", + local_schema_hash.to_rihs_string(), + // Render both sides through an ungated formatter. `ti.hash` is a + // `hiroz_protocol::TypeHash`, whose `to_rihs_string` is + // `#[cfg(feature = "no-type-hash")]`-gated: under that feature it + // returns the constant "TypeHashNotSupported" for every value, so + // this message would tell the reader the publisher advertises + // nothing while we refuse *because* it advertises something else. + rihs_ungated(&ti.hash) + )); + } + HashCheck::Match => {} + } + + // Order matters: `with_type_info` assigns unconditionally, so it must follow + // `create_dyn_sub` (which recomputes the hash from the local .msg). With the + // hashes now proven equal this is belt-and-braces, and it keeps the key + // byte-identical to what `create_dyn_sub_auto` would have declared. + let sub = node + .create_dyn_sub(topic, schema) + .with_type_info(ti) + .build() + .map_err(|e| { + format!("declaring a dynamic subscriber for {topic} ({canonical}) failed: {e}") + })?; + + tracing::info!("resolved schema for {topic} from a local .msg ({canonical})"); + Ok(sub) +} + /// The longest time that a guest-supplied service timeout may suspend the /// epoch ticker. /// @@ -46,24 +195,11 @@ fn clamp_guest_timeout(timeout_ms: u32) -> Duration { } impl PluginState { - /// The message type advertised by a live publisher or subscriber on `topic`, - /// if any. Used both to build the concrete publish key (`resolve_topic_ke`) - /// and to reject a disk-resolved type that conflicts with what the topic - /// actually carries (`encode_yaml_to_cdr`). + /// See [`live_topic_type_info`]. Used both to build the concrete publish key + /// (`resolve_topic_ke`) and to reject a disk-resolved type that conflicts + /// with what the topic actually carries (`encode_yaml_to_cdr`). fn live_topic_type_info(&self, topic: &str) -> Option { - use hiroz_protocol::{EndpointKind, Entity}; - [EndpointKind::Publisher, EndpointKind::Subscription] - .into_iter() - .find_map(|kind| { - self.engine - .graph - .get_entities_by_topic(kind, topic) - .first() - .and_then(|ent| match ent.as_ref() { - Entity::Endpoint(ep) => ep.type_info.clone(), - _ => None, - }) - }) + live_topic_type_info(&self.engine.graph, topic) } } @@ -89,26 +225,60 @@ impl hu::plugin::ros::Host for PluginState { topic: String, ) -> Result, PluginError> { self.require_perm(hu::plugin::types::Permission::SubscribeTopic)?; - let rep = self.alloc_rep(); - let (tx, rx) = flume::bounded::(256); + // Resolve the schema *before* returning a Subscription. Inside the spawned + // task every failure was invisible: `subscribe` had already returned Ok, + // so the plugin's error branch could never fire and a dropped sender read + // as a permanently idle topic. + // + // `block_in_place` hands the worker to the blocking pool, so the zenoh + // I/O this discovery depends on keeps progressing. `HostBlockGuard` stops + // the epoch ticker: the guest runs under a ~3 s wall-clock budget, so + // without it a long wait traps the guest before it can run that error + // branch, and the command hangs instead of reporting. + const SUB_DISCOVERY_TIMEOUT: Duration = Duration::from_secs(5); let node = self.engine.node.clone(); - let topic_clone = topic.clone(); + let discovered = { + let _epoch = super::super::HostBlockGuard::enter(); + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current() + .block_on(node.create_dyn_sub_auto(&topic, SUB_DISCOVERY_TIMEOUT)) + }) + }; + let sub = match discovered { + Ok(sub) => sub, + Err(e) => { + // Discovery is authoritative when it answers, but a node built + // without `.with_type_description_service()` answers nothing -- + // so fall back to the same on-disk `.msg` lookup the publish + // path uses (`encode_yaml_to_cdr`). The type *name* still comes + // from the graph; only the schema body comes from disk. + tracing::debug!("WASM plugin: schema discovery failed for {topic}: {e}"); + match dyn_sub_from_local_msg(&node, &self.engine.graph, &topic) { + Ok(sub) => sub, + // Both reasons are reported: which one matters depends on + // whether the user expected discovery or the disk to answer. + Err(fallback) => { + return Err(PluginError::Transport(format!( + "no schema for {topic}: discovery failed ({e}); {fallback}" + ))); + } + } + } + }; + + // Mint the resource only once the subscription really exists, so a + // failed subscribe doesn't burn a rep. + let rep = self.alloc_rep(); + let (tx, rx) = flume::bounded::(256); + let topic_for_log = topic.clone(); let handle = tokio::spawn(async move { - let sub = match node - .create_dyn_sub_auto(&topic_clone, Duration::from_secs(5)) - .await - { - Ok(s) => s, - Err(e) => { - tracing::warn!( - "WASM plugin: schema discovery failed for {}: {e}", - topic_clone - ); - return; - } - }; + // A stream that decodes to nothing is indistinguishable from an idle + // one, which is the same silence this whole path had. Report the + // first failure loudly and the rest at debug, so a fast topic whose + // schema drifted announces itself without flooding stderr. + let mut decode_errors: u64 = 0; loop { match sub.try_recv() { Some(Ok(msg)) => { @@ -117,7 +287,22 @@ impl hu::plugin::ros::Host for PluginState { break; } } - Some(Err(_)) => {} + Some(Err(e)) => { + decode_errors += 1; + if decode_errors == 1 { + tracing::warn!( + topic = %topic_for_log, + "WASM plugin: message decode failed: {e} \ + (further errors logged at debug)" + ); + } else { + tracing::debug!( + topic = %topic_for_log, + count = decode_errors, + "WASM plugin: message decode failed: {e}" + ); + } + } None => { tokio::time::sleep(Duration::from_millis(5)).await; } @@ -535,6 +720,33 @@ fn json_to_dynamic_value( } } +#[cfg(test)] +mod graph_type_name_tests { + use hiroz::dynamic::ros_type_name_from_dds; + + // `dyn_sub_from_local_msg` feeds a graph-reported (DDS-mangled) type name + // straight into `load_schema`, which only accepts the canonical form. Pin the + // conversion at the crate boundary -- a regression here shows up as "no .msg + // found" rather than as a type error. + #[test] + fn graph_type_names_normalize_for_schema_lookup() { + assert_eq!( + ros_type_name_from_dds("std_msgs::msg::dds_::String_"), + "std_msgs/msg/String" + ); + // Some publishers report the un-`dds_`-qualified form. + assert_eq!( + ros_type_name_from_dds("rcl_interfaces::msg::ParameterEvent_"), + "rcl_interfaces/msg/ParameterEvent" + ); + // Already-canonical names must survive unchanged. + assert_eq!( + ros_type_name_from_dds("std_msgs/msg/String"), + "std_msgs/msg/String" + ); + } +} + #[cfg(test)] mod service_type_name_tests { use super::service_request_response_type_names; @@ -586,6 +798,76 @@ mod service_type_name_tests { } } +#[cfg(test)] +mod type_hash_guard_tests { + use super::{HashCheck, check_type_hash, rihs_ungated}; + use hiroz::TypeHash; + + fn hash(byte: u8) -> TypeHash { + TypeHash::new(1, [byte; 32]) + } + + #[test] + fn equal_hashes_match() { + assert_eq!(check_type_hash(&hash(0xab), &hash(0xab)), HashCheck::Match); + } + + #[test] + fn different_hashes_mismatch() { + assert_eq!( + check_type_hash(&hash(0xab), &hash(0xcd)), + HashCheck::Mismatch + ); + } + + // The case the guard got wrong in the other direction: a peer that + // advertises nothing is unverifiable, not mismatched. Refusing it made every + // cross-distro subscribe fail, which is the headline case for a disk + // fallback existing at all. + #[test] + fn absent_advertised_hash_is_not_a_mismatch() { + assert_eq!( + check_type_hash(&hash(0xab), &TypeHash::zero()), + HashCheck::NotAdvertised + ); + } + + // A local .msg cannot hash to zero, but pin the precedence anyway: the + // "not advertised" arm is checked first, so two zeros are not a match. + #[test] + fn absent_beats_equality_when_both_are_zero() { + assert_eq!( + check_type_hash(&TypeHash::zero(), &TypeHash::zero()), + HashCheck::NotAdvertised + ); + } + + // The reason the comparison is by value and not by rendered string. Under + // `no-type-hash` the protocol renderer returns one constant for every hash, + // so any string comparison passes on anything. This asserts the property + // that made string comparison wrong, and it holds on every build. + #[test] + fn distinct_hashes_stay_distinct_by_value() { + let a = hash(0x01); + let b = hash(0x02); + assert_ne!(a, b); + assert_eq!(check_type_hash(&a, &b), HashCheck::Mismatch); + } + + // The diagnostic must show the advertised bytes. `to_rihs_string` is gated + // and would print "TypeHashNotSupported" under `no-type-hash`, telling the + // reader the publisher advertised nothing while we refuse because it + // advertised something else. + #[test] + fn ungated_renderer_shows_the_bytes() { + assert_eq!( + rihs_ungated(&hash(0xab)), + format!("RIHS01_{}", "ab".repeat(32)) + ); + assert_ne!(rihs_ungated(&hash(0xab)), "TypeHashNotSupported"); + } +} + #[cfg(test)] mod guest_timeout_tests { use super::{MAX_GUEST_TIMEOUT, clamp_guest_timeout};