diff --git a/crates/hiroz-tests/tests/message_lost.rs b/crates/hiroz-tests/tests/message_lost.rs new file mode 100644 index 00000000..621d7f30 --- /dev/null +++ b/crates/hiroz-tests/tests/message_lost.rs @@ -0,0 +1,186 @@ +//! `RMW_EVENT_MESSAGE_LOST` must actually be raised by a live subscriber. +//! +//! `event.rs`'s unit tests pin `MessageLossTracker`'s arithmetic. They say +//! nothing about whether anything *calls* it: delete the `observe_loss(..)` line +//! from the subscriber's receive path and every one of them still passes. This +//! file is the detector for that wiring. +//! +//! Loss is induced deterministically rather than by trying to drop a packet. +//! The subscriber's own key expression is published to directly, through the +//! node's zenoh session, with a hand-built [`Attachment`] carrying a chosen +//! sequence number — so the gap is exact and there is no timing to lose. + +mod common; + +use std::{ + sync::{Arc, Mutex}, + thread, + time::{Duration, Instant}, +}; + +use common::{TestRouter, create_hiroz_context_with_endpoint}; +use hiroz::{ + Builder, GidArray, TypeHash, attachment::Attachment, event::ZenohEventType, + ros_msg::MessageTypeInfo, +}; +use serde::{Deserialize, Serialize}; +use serial_test::serial; +use zenoh::Wait; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +struct Tick { + counter: u64, +} + +impl MessageTypeInfo for Tick { + fn type_name() -> &'static str { + "test_msgs::msg::dds_::Tick_" + } + fn type_hash() -> TypeHash { + TypeHash::zero() + } +} + +impl hiroz::ros_msg::WithTypeInfo for Tick {} + +impl hiroz::msg::ZMessage for Tick { + type Serdes = hiroz::msg::SerdeCdrSerdes; +} + +fn gid(n: u8) -> GidArray { + let mut g = [0u8; 16]; + g[0] = n; + g +} + +/// A CDR-encoded `Tick`, ready to hand to `Session::put`. +fn payload(counter: u64) -> zenoh::bytes::ZBytes { + use hiroz::msg::ZSerializer; + let zbuf = >::serialize_to_zbuf(&Tick { counter }); + zenoh::bytes::ZBytes::from(zbuf) +} + +/// Wait until `total_count` for `MessageLost` stops changing, then return it. +fn settled_loss_count(sub_events: &Arc>) -> i32 { + let mut last = -1; + let mut stable_since = Instant::now(); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let now = sub_events + .lock() + .unwrap() + .take_event_status(ZenohEventType::MessageLost) + .total_count; + if now != last { + last = now; + stable_since = Instant::now(); + } else if stable_since.elapsed() >= Duration::from_millis(400) { + return now; + } + assert!(Instant::now() < deadline, "loss count never settled"); + thread::sleep(Duration::from_millis(25)); + } +} + +/// A gap in a publisher's sequence numbers raises `MessageLost` on the +/// subscriber that saw it, with the count of samples that never arrived. +#[test] +#[serial] +fn a_sequence_gap_raises_message_lost() { + const TOPIC: &str = "/message_lost_gap"; + + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("context"); + let node = ctx.create_node("message_lost_node").build().expect("node"); + + let received = Arc::new(Mutex::new(Vec::::new())); + let cb_received = received.clone(); + let sub = node + .create_sub::(TOPIC) + .build_with_callback(move |msg: Tick| { + cb_received.lock().unwrap().push(msg.counter); + }) + .expect("subscriber"); + + // Publish straight onto the subscriber's own key expression, so the + // sequence numbers are ours to choose. + let ke = node + .keyexpr_format() + .topic_key_expr(sub.entity()) + .expect("topic key expr"); + let session = node.session(); + let publisher_gid = gid(42); + + let put = |sn: i64, counter: u64| { + session + .put((*ke).clone(), payload(counter)) + .attachment(Attachment::new(sn, publisher_gid)) + .wait() + .expect("put"); + }; + + thread::sleep(Duration::from_millis(300)); + + put(0, 0); // baseline — first from this publisher, never counted + put(1, 1); // contiguous + put(5, 5); // 2, 3 and 4 never arrived + + let lost = settled_loss_count(sub.events_mgr()); + + assert_eq!( + lost, 3, + "expected the three skipped sequence numbers to be reported as lost; \ + got {lost}. Zero means the receive path never fed the loss tracker" + ); + assert_eq!( + received.lock().unwrap().len(), + 3, + "all three published samples should still have been delivered — \ + detecting loss must not drop anything" + ); +} + +/// A subscriber that joins late has not "lost" the history it was never sent. +/// +/// Without the first-sample exemption this reports the publisher's sequence +/// number as the loss count, so every late joiner looks catastrophically lossy. +#[test] +#[serial] +fn joining_late_reports_no_loss() { + const TOPIC: &str = "/message_lost_late_join"; + + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("context"); + let node = ctx + .create_node("message_lost_late_node") + .build() + .expect("node"); + + let sub = node + .create_sub::(TOPIC) + .build_with_callback(|_msg: Tick| {}) + .expect("subscriber"); + + let ke = node + .keyexpr_format() + .topic_key_expr(sub.entity()) + .expect("topic key expr"); + let session = node.session(); + + thread::sleep(Duration::from_millis(300)); + + // First sample this subscriber ever sees from this publisher, and it is + // already well into the publisher's stream. + session + .put((*ke).clone(), payload(9000)) + .attachment(Attachment::new(9000, gid(7))) + .wait() + .expect("put"); + + let lost = settled_loss_count(sub.events_mgr()); + + assert_eq!( + lost, 0, + "a late joiner must not be charged for history it was never sent" + ); +} diff --git a/crates/hiroz/src/event.rs b/crates/hiroz/src/event.rs index 9dea9e7a..e7b364bb 100644 --- a/crates/hiroz/src/event.rs +++ b/crates/hiroz/src/event.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -445,6 +446,87 @@ pub fn update_shared_event_status( update_shared_event_status_with_policy(events_mgr, event_type, change, 0) } +/// Detects samples lost **in transit** and raises [`ZenohEventType::MessageLost`]. +/// +/// Every sample carries an [`Attachment`] with the publisher's GID and a +/// per-publisher sequence number. Holding the last sequence seen from each +/// publisher makes a gap detectable: receiving `n` when `n - 2` was the last +/// means one sample never arrived. +/// +/// [`Attachment`]: crate::attachment::Attachment +/// +/// # What this does *not* count +/// +/// A subscriber dropping its own oldest queued sample because the queue is at +/// its history depth. That sample **arrived** — it updated the last-seen +/// sequence on the way in — so it produces no gap, and the ROS event does not +/// claim it. `rmw_zenoh_cpp` draws the line in the same place: its depth-drops +/// are a debug log, and only sequence gaps raise `MESSAGE_LOST`. +pub struct MessageLossTracker { + events_mgr: Arc>, + /// Last sequence number seen per publisher GID. + /// + /// Its own lock, and never held across the callout below — raising the + /// event runs user code, which may re-enter this subscriber. + last_seen: Mutex>, +} + +impl MessageLossTracker { + pub fn new(events_mgr: Arc>) -> Self { + Self { + events_mgr, + last_seen: Mutex::new(HashMap::new()), + } + } + + /// Record an arrival, raising the event if it skipped past anything. + pub fn observe(&self, source_gid: GidArray, sequence_number: i64) { + let lost = { + let Ok(mut seen) = self.last_seen.lock() else { + return; + }; + match seen.entry(source_gid) { + // First sample from this publisher. There is no baseline to + // measure against, and a subscriber that joined late has not + // "lost" the history it was never sent. + Entry::Vacant(slot) => { + slot.insert(sequence_number); + 0 + } + Entry::Occupied(mut slot) => { + let high_water = *slot.get(); + // The baseline only ever moves **forward**. An arrival at or + // below it is a replay, a retransmit or a reorder — not + // loss — and letting it move the baseline backwards would + // make the *next* ordinary sample look like a gap. + // + // This is a deliberate divergence from `rmw_zenoh_cpp`, + // which uses `std::abs(sn - last)` and rewrites the + // baseline unconditionally. On a `TransientLocal` + // subscriber, history replay delivers older sequence + // numbers as a matter of course, so that shape reports + // phantom loss twice per replayed sample. + if sequence_number <= high_water { + 0 + } else { + slot.insert(sequence_number); + sequence_number.saturating_sub(high_water).saturating_sub(1) + } + } + } + }; + + if lost > 0 { + // Clamped rather than truncated: the rmw status field is i32, and a + // publisher that restarts its numbering can present an arbitrarily + // large apparent jump. (In ROS a restarted endpoint normally gets a + // fresh GID and lands in the vacant arm instead.) + let lost = lost.min(i64::from(i32::MAX)) as i32; + update_shared_event_status(&self.events_mgr, ZenohEventType::MessageLost, lost); + } + } +} + /// [`update_shared_event_status`] with a QoS policy kind. /// /// # Known hazard @@ -556,6 +638,65 @@ mod tests { assert_eq!(status.current_count, 0); } + /// Drive a tracker through a sequence of arrivals and return the total + /// `MessageLost` count it reported. + fn losses_for(arrivals: &[(u8, i64)]) -> i32 { + let mgr = Arc::new(Mutex::new(EventsManager::new(gid(1)))); + let tracker = MessageLossTracker::new(mgr.clone()); + for &(publisher, sn) in arrivals { + tracker.observe(gid(publisher), sn); + } + mgr.lock() + .unwrap() + .take_event_status(ZenohEventType::MessageLost) + .total_count + } + + #[test] + fn message_loss_is_counted_from_sequence_gaps() { + // 0,1,2 contiguous → nothing lost. Then 5 skips 3 and 4. + assert_eq!(losses_for(&[(1, 0), (1, 1), (1, 2), (1, 5)]), 2); + } + + #[test] + fn message_loss_ignores_the_first_sample_from_a_publisher() { + // A late joiner's first sample has no baseline. Reporting `sn` as the + // loss count would make every subscriber that starts late look lossy. + assert_eq!(losses_for(&[(1, 9_000)]), 0); + } + + #[test] + fn message_loss_is_tracked_per_publisher() { + // Interleaved publishers each keep their own baseline; without that, + // alternating 0,0,1,1 reads as a gap on every other sample. + assert_eq!(losses_for(&[(1, 0), (2, 0), (1, 1), (2, 1)]), 0); + } + + #[test] + fn message_loss_ignores_reorder_and_republish() { + // A non-positive difference is a retransmit, a reorder, or a publisher + // that restarted its numbering — none of which is loss. + assert_eq!(losses_for(&[(1, 5), (1, 3), (1, 5), (1, 0)]), 0); + } + + /// A replayed sample must not make the *next* ordinary one look like a gap. + /// + /// This is the case `rmw_zenoh_cpp` gets wrong: `std::abs(sn - last)` plus + /// an unconditional baseline rewrite reports 1 lost for the replay and 2 + /// more for the sample after it. Every `TransientLocal` subscriber replays + /// history, so it is reachable rather than theoretical. + #[test] + fn message_loss_survives_a_transient_local_replay() { + assert_eq!(losses_for(&[(1, 5), (1, 3), (1, 6)]), 0); + } + + #[test] + fn message_loss_clamps_an_implausible_jump() { + // A restarted publisher can present an arbitrarily large apparent gap; + // the rmw status field is i32, so it must saturate rather than wrap. + assert_eq!(losses_for(&[(1, 0), (1, i64::MAX)]), i32::MAX); + } + #[test] fn test_update_event_status_fires_callback() { let called = Arc::new(Mutex::new(0i32)); diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 5b637d03..b03bb30b 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -10,7 +10,7 @@ use crate::Builder; use crate::attachment::{Attachment, GidArray}; use crate::common::DataHandler; use crate::entity::{EndpointEntity, EndpointKind}; -use crate::event::EventsManager; +use crate::event::{EventsManager, MessageLossTracker}; use crate::graph::Graph; use crate::impl_with_type_info; use crate::queue::BoundedQueue; @@ -36,6 +36,22 @@ const SAMPLE_MISS_HEARTBEAT_PERIOD: Duration = Duration::from_millis(500); /// truncation surprises inside zenoh-ext's internal `as_millis()` paths). const TRANSIENT_LOCAL_QUERY_TIMEOUT: Duration = Duration::from_millis(u64::MAX); +/// Feed an arriving sample's attachment to the loss tracker. +/// +/// A sample whose attachment is missing or undecodable is **ignored, not +/// counted**. The sequence number is the only thing that makes loss detectable, +/// and a publisher that does not send one — a plain zenoh peer rather than a +/// hiroz node — must not be reported as lossy just for being unrecognised. +fn observe_loss(loss: &MessageLossTracker, sample: &Sample) { + let Some(bytes) = sample.attachment() else { + return; + }; + let Ok(attachment) = Attachment::try_from(bytes) else { + return; + }; + loss.observe(attachment.source_gid, attachment.sequence_number); +} + fn cache_depth_from_history(history: QosHistory) -> usize { // Mirrors rmw_zenoh_cpp's `QoS::best_available_qos` (`qos.cpp:107`): // a zero-valued depth (the rmw representation of "unspecified" or @@ -802,9 +818,17 @@ where key_expr, self.entity.qos ); + // Built here rather than at `ZSub` construction: the loss tracker needs + // a handle to it, and the tracker is captured by the callback below. + let gid = crate::entity::endpoint_gid(&self.entity) + .expect("local endpoint always has node identity"); + let events_mgr = Arc::new(Mutex::new(EventsManager::new(gid))); + let loss = Arc::new(MessageLossTracker::new(events_mgr.clone())); + // Wrap handler with encoding validation if expected encoding is set let expected_encoding = self.expected_encoding.clone(); let validated_handler = move |sample: Sample| { + observe_loss(&loss, &sample); // Validate encoding if expected encoding is set if let Some(ref expected) = expected_encoding { let encoding_str = sample.encoding().to_string(); @@ -841,8 +865,6 @@ where let sub_builder = apply_transient_local_sub(sub_builder, &self.entity.qos); let inner = sub_builder.wait()?; - let gid = crate::entity::endpoint_gid(&self.entity) - .expect("local endpoint always has node identity"); let lv_ke = self .keyexpr_format .liveliness_key_expr(&self.entity, &self.session.zid())?; @@ -859,7 +881,7 @@ where _inner: inner, _lv_token: lv_token, queue, - events_mgr: Arc::new(Mutex::new(EventsManager::new(gid))), + events_mgr, graph: self.graph, dyn_schema: self.dyn_schema, expected_encoding: self.expected_encoding, @@ -990,6 +1012,25 @@ impl std::fmt::Debug for ZSub { } } +/// Accessors that do not depend on how the subscriber delivers. +/// +/// These were on the `ZSub` (queue-mode) impl only, so a callback +/// subscriber could not reach its own events manager — the handle the rmw layer +/// needs to install an event callback, and the only way to observe +/// [`MessageLost`](crate::event::ZenohEventType::MessageLost). Neither field has +/// anything to do with the queue. +impl ZSub { + /// The event manager this subscriber raises endpoint events on. + pub fn events_mgr(&self) -> &Arc> { + &self.events_mgr + } + + /// Get a reference to the endpoint entity for this subscriber. + pub fn entity(&self) -> &EndpointEntity { + &self.entity + } +} + impl ZSub where T: ZMessage, @@ -1021,15 +1062,6 @@ where .ok_or_else(|| crate::error::Error::timeout(timeout)) } - pub fn events_mgr(&self) -> &Arc> { - &self.events_mgr - } - - /// Get a reference to the endpoint entity for this subscriber. - pub fn entity(&self) -> &EndpointEntity { - &self.entity - } - /// Check if there are messages available in the queue pub fn is_ready(&self) -> bool { self.queue.as_ref().map(|q| !q.is_empty()).unwrap_or(false)