fix(rmw): borrow guard conditions immutably, tighten the graph-event assertion

**Soundness.** `rmw_trigger_guard_condition` took `&mut GuardConditionImpl`
while `rmw_wait` holds `&GuardConditionImpl` to the same object. Moving
`triggered` behind an `AtomicBool` defined the data race but said nothing
about the aliasing: handing out a `&mut` to an object another thread holds a
`&` to is undefined behaviour whatever the field types are. All mutation
already goes through atomics in `GuardConditionState`, so `trigger` and
`reset` now take `&self`, the FFI entry point borrows via `borrow_data`, and
the wait-set accessor takes a shared reference.

**Test strength.** `reentrant_graph_event.rs` asserted the re-entrant query
saw `>= 1` publisher -- satisfiable by `local_pub`, which exists for the whole
scenario. A regression invoking the callback *before* inserting the remote
entity would still return 1 and pass, so the assertion did not depend on the
ordering it claimed to verify. It now requires both.

**Lock scope.** `EventsManager::set_callback` fires the backlog callback while
the caller's outer `Mutex<EventsManager>` guard is live -- the "outside the
lock" it releases is only the inner `event_mutex`. The docs now say so, and
`set_shared_callback` is the safe entry point for `&Mutex<EventsManager>`
holders, mirroring `update_shared_event_status`. No caller needed migrating:
`RmEventHandle::set_callback` already collects under the guard and fires after
releasing, and the remaining call sites own their manager outright.

Author: YuanYuYuan

crates/hiroz-tests/tests/reentrant_graph_event.rs

@@ -292,10 +292,18 @@ fn graph_change_callback_querying_the_graph_does_not_deadlock() {
             "the graph-change callback never ran for the remote publisher — \
              the scenario proved nothing"
         );
+        // Two, not one. `local_pub` is on this node for the whole scenario, so
+        // `>= 1` was satisfiable without the remote publisher ever being in the
+        // graph — which is precisely the regression this is meant to catch: a
+        // callback invoked *before* the entity is inserted would still observe
+        // the local publisher and pass. Requiring both makes the assertion
+        // actually depend on the ordering it claims to verify.
         assert!(
-            counted.load(Ordering::SeqCst) >= 1,
-            "the re-entrant graph query returned {} publishers; it should see at \
-             least the remote one",
+            counted.load(Ordering::SeqCst) >= 2,
+            "the re-entrant graph query saw {} publisher(s) on the topic; it must \
+             see both the local one and the remote one whose appearance triggered \
+             this callback. Seeing exactly 1 means the callback ran before the \
+             remote entity was inserted into the graph",
             counted.load(Ordering::SeqCst)
         );
     });

crates/hiroz/src/event.rs

@@ -69,14 +69,25 @@ impl EventsManager {
         }
     }
 
+    /// Install a callback, delivering any backlog immediately.
+    ///
+    /// **Only for a caller that owns this manager outright.** `&mut self` means
+    /// the caller holds whatever `Mutex<EventsManager>` wraps it, and the
+    /// backlog callback below runs while that outer guard is still live — so a
+    /// callback that re-enters (`RmEventHandle::take_event`, say) deadlocks on
+    /// it. The "outside the lock" this releases is only the *inner*
+    /// `event_mutex`.
+    ///
+    /// Anyone holding an `Arc<Mutex<EventsManager>>` must use
+    /// [`set_shared_callback`] instead, which is to registration what
+    /// [`update_shared_event_status`] is to status updates.
     pub fn set_callback<F>(&mut self, event_type: ZenohEventType, callback: F)
     where
         F: Fn(i32) + Send + Sync + 'static,
     {
         let callback: EventCallback = Arc::new(callback);
         let unread_count = self.install_callback(event_type, callback.clone());
-        // If there were unread events, notify the freshly-registered callback —
-        // outside the lock.
+        // Outside the inner `event_mutex` only — see the note above.
         if unread_count != 0 {
             callback(unread_count);
         }
@@ -427,6 +438,40 @@ impl EventWaitData {
 /// outer guard alive across the callback, and the callback is user code handed
 /// to an rclcpp executor which routinely calls straight back into the same
 /// manager (`rmw_take_event` → [`RmEventHandle::take_event`]).
+/// Install a callback on a shared manager, delivering any backlog **after**
+/// the outer guard is released.
+///
+/// The registration counterpart of [`update_shared_event_status`], and the
+/// entry point every holder of an `Arc<Mutex<EventsManager>>` must use.
+/// [`EventsManager::set_callback`] takes `&mut self`, so it can only be called
+/// with the outer mutex already held, and it fires the backlog underneath it —
+/// a callback that re-enters (`RmEventHandle::take_event`) then self-deadlocks
+/// on a non-reentrant `Mutex`. This collects the backlog under the guard, drops
+/// it, and only then calls.
+pub fn set_shared_callback<F>(
+    events_mgr: &Mutex<EventsManager>,
+    event_type: ZenohEventType,
+    callback: F,
+) where
+    F: Fn(i32) + Send + Sync + 'static,
+{
+    let callback: EventCallback = Arc::new(callback);
+
+    // Bound to its own `let` inside a block, for the same reason as
+    // `update_shared_event_status_with_policy`: as a `match`/`if let` scrutinee
+    // the guard would outlive the invocation below and reinstate the deadlock.
+    let unread_count = {
+        let Ok(mut mgr) = events_mgr.lock() else {
+            return;
+        };
+        mgr.install_callback(event_type, callback.clone())
+    };
+
+    if unread_count != 0 {
+        callback(unread_count);
+    }
+}
+
 pub fn update_shared_event_status(
     events_mgr: &Mutex<EventsManager>,
     event_type: ZenohEventType,

crates/rmw-zenoh-rs/src/guard_condition.rs

@@ -55,11 +55,19 @@ pub struct GuardConditionImpl {
 }
 
 impl GuardConditionImpl {
-    pub(crate) fn trigger(&mut self) -> Result<(), ()> {
+    // `&self`, not `&mut self`. `rmw_wait` holds shared references to this
+    // object while scanning the wait set, and `rmw_trigger_guard_condition` can
+    // fire from a zenoh delivery thread at the same time. Handing out a `&mut`
+    // to an object another thread holds a `&` to is undefined behaviour in Rust
+    // whatever the field types are -- moving `triggered` behind an atomic
+    // defines the *data race* but says nothing about the aliasing. All mutation
+    // now goes through the atomics in `GuardConditionState`, so shared access is
+    // sufficient and every borrow can be immutable.
+    pub(crate) fn trigger(&self) -> Result<(), ()> {
         self.state.fire()
     }
 
-    pub fn reset(&mut self) {
+    pub fn reset(&self) {
         self.state.reset();
     }
 
@@ -138,7 +146,8 @@ pub extern "C" fn rmw_trigger_guard_condition(
         return RMW_RET_INVALID_ARGUMENT as _;
     }
 
-    if let Ok(gc_impl) = (guard_condition as *mut rmw_guard_condition_t).borrow_mut_data() {
+    // Immutable borrow: see the note on `GuardConditionImpl::trigger`.
+    if let Ok(gc_impl) = (guard_condition as *mut rmw_guard_condition_t).borrow_data() {
         let _ = gc_impl.trigger();
     }
 

crates/rmw-zenoh-rs/src/wait_set.rs

@@ -388,8 +388,10 @@ pub extern "C" fn rmw_wait(
                     unsafe { *gc_array.guard_conditions.add(i) as *mut rmw_guard_condition_impl_t };
                 if !gc_impl_ptr.is_null() {
                     unsafe {
+                        // Shared, not exclusive: a delivery thread may be
+                        // inside `trigger` on this same object right now.
                         let gc_impl =
-                            &mut *(gc_impl_ptr as *mut crate::guard_condition::GuardConditionImpl);
+                            &*(gc_impl_ptr as *const crate::guard_condition::GuardConditionImpl);
                         if !gc_impl.is_ready() {
                             // Not ready - set to NULL in place
                             *gc_array.guard_conditions.add(i) = std::ptr::null_mut();