feat(rmw-zenoh-rs): add lock-tripwire as a regression guard

crates/rmw-zenoh-rs/Cargo.toml

@@ -36,6 +36,7 @@ hiroz = { workspace = true, features = ["rmw"] }
 hiroz-protocol = { workspace = true }
 hiroz-schema = { workspace = true }
 strum = { workspace = true, features = ["derive"] }
+lock-tripwire = { git = "https://github.com/YuanYuYuan/lock-tripwire.git", tag = "v0.1.1", optional = true }
 
 [lib]
 name = "rmw_zenoh_rs"
@@ -58,5 +59,16 @@ test-all = ["test-core", "test-msgs"]
 # Legacy compatibility
 test-rmw = ["test-all"]
 
+# Wires lock-tripwire into the callback-holder/user_data Mutex fields
+# (Subscription/Service/Client). Off by default -- no change to shipped
+# behavior. The plain fix already removes the deadlock this guards
+# against; this is a regression guard, not a live fix -- it fires only if
+# a future change reintroduces holding one of these locks across a
+# call-out. Live in cargo test / a plain debug build; a no-op in a plain
+# --release build unless lock-tripwire's own force-checks feature (or its
+# per-package profile override) is also enabled -- see
+# tripwire_compat.rs's module docs.
+lock-tripwire-guard = ["dep:lock-tripwire"]
+
 [dev-dependencies]
 pyo3 = { version = "0.22", features = ["auto-initialize"] }

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

@@ -23,6 +23,7 @@ pub mod ros;
 pub mod service;
 #[macro_use]
 pub mod traits;
+pub mod tripwire_compat;
 pub mod type_support;
 pub mod utils;
 pub mod wait_set;

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

@@ -83,27 +83,40 @@ impl PublisherImpl {
 pub(crate) fn build_subscription_notify_callback(
     notifier: std::sync::Arc<crate::utils::Notifier>,
     callback_holder: std::sync::Arc<
-        std::sync::Mutex<crate::ros::rmw_subscription_new_message_callback_t>,
+        crate::tripwire_compat::GuardedMutex<crate::ros::rmw_subscription_new_message_callback_t>,
     >,
-    user_data_holder: std::sync::Arc<std::sync::Mutex<usize>>,
+    user_data_holder: std::sync::Arc<crate::tripwire_compat::GuardedMutex<usize>>,
     unread_count_holder: std::sync::Arc<std::sync::Mutex<usize>>,
 ) -> impl Fn() + Send + Sync + 'static {
+    use crate::tripwire_compat::LockGuarded;
     move || {
         notifier.notify_all();
-        // The `.lock()` temporary is dropped at the end of this statement --
-        // released before any call-out below, unlike a `match`/`if let`
-        // scrutinee, which would extend it across the whole arm.
-        let Ok(callback_fn) = callback_holder.lock().map(|g| *g) else {
+        // The `.lock_guarded()` temporary is dropped at the end of this
+        // statement -- released before any call-out below, unlike a
+        // `match`/`if let` scrutinee, which would extend it across the
+        // whole arm.
+        let Ok(callback_fn) = callback_holder
+            .lock_guarded("rmw_zenoh_rs::pubsub::notify_callback::callback")
+            .map(|g| *g)
+        else {
             return;
         };
         match callback_fn {
             Some(callback_fn) => {
                 // Copied out and the lock released before the call-out --
                 // the setter locks this same mutex, so holding it here
                 // would be a second AB-BA pair alongside `callback_holder`.
-                if let Ok(user_data_usize) = user_data_holder.lock().map(|g| *g) {
+                if let Ok(user_data_usize) = user_data_holder
+                    .lock_guarded("rmw_zenoh_rs::pubsub::notify_callback::user_data")
+                    .map(|g| *g)
+                {
                     let user_data_ptr = user_data_usize as *const std::ffi::c_void;
-                    unsafe { callback_fn(user_data_ptr, 1) }; // 1 new message
+                    // Both locks above are already released -- this asserts
+                    // that stays true even if a future change regresses it.
+                    crate::guarded_call!(
+                        "rmw_zenoh_rs::pubsub::notify_callback::call_out",
+                        unsafe { callback_fn(user_data_ptr, 1) } // 1 new message
+                    );
                 }
             }
             None => {
@@ -122,13 +135,19 @@ pub(crate) fn build_subscription_notify_callback(
 /// stores the new callback, then calls out -- all three locks released
 /// before the call, none held during it.
 pub(crate) fn set_subscription_callback_core(
-    callback_holder: &std::sync::Mutex<crate::ros::rmw_subscription_new_message_callback_t>,
-    user_data_holder: &std::sync::Mutex<usize>,
+    callback_holder: &crate::tripwire_compat::GuardedMutex<
+        crate::ros::rmw_subscription_new_message_callback_t,
+    >,
+    user_data_holder: &crate::tripwire_compat::GuardedMutex<usize>,
     unread_count_holder: &std::sync::Mutex<usize>,
     callback: crate::ros::rmw_subscription_new_message_callback_t,
     user_data: *mut crate::c_void,
 ) {
-    if let Ok(mut ud) = user_data_holder.lock() {
+    use crate::tripwire_compat::LockGuarded;
+
+    if let Ok(mut ud) = user_data_holder
+        .lock_guarded("rmw_zenoh_rs::rmw_subscription_set_on_new_message_callback::user_data")
+    {
         *ud = user_data as usize;
     }
 
@@ -138,7 +157,9 @@ pub(crate) fn set_subscription_callback_core(
     // independently of this lock would make a poisoned callback_holder
     // silently reset progress and still fire the call, which is a real
     // (if narrow) behavior change from before, not just a refactor.
-    let Ok(mut cb) = callback_holder.lock() else {
+    let Ok(mut cb) = callback_holder
+        .lock_guarded("rmw_zenoh_rs::rmw_subscription_set_on_new_message_callback::callback")
+    else {
         return;
     };
     let pending = if callback.is_some() {
@@ -155,7 +176,10 @@ pub(crate) fn set_subscription_callback_core(
 
     if let (Some(callback_fn), Some(n)) = (callback, pending) {
         if n > 0 {
-            unsafe { callback_fn(user_data as *const std::ffi::c_void, n) };
+            crate::guarded_call!(
+                "rmw_zenoh_rs::rmw_subscription_set_on_new_message_callback::call_out",
+                unsafe { callback_fn(user_data as *const std::ffi::c_void, n) }
+            );
         }
     }
 }
@@ -167,9 +191,10 @@ pub struct SubscriptionImpl {
     pub topic: CString,
     pub options: rmw_subscription_options_t,
     pub qos: rmw_qos_profile_t,
-    pub callback:
-        std::sync::Arc<std::sync::Mutex<crate::ros::rmw_subscription_new_message_callback_t>>,
-    pub callback_user_data: std::sync::Arc<std::sync::Mutex<usize>>, // Store pointer as usize for thread safety
+    pub callback: std::sync::Arc<
+        crate::tripwire_compat::GuardedMutex<crate::ros::rmw_subscription_new_message_callback_t>,
+    >,
+    pub callback_user_data: std::sync::Arc<crate::tripwire_compat::GuardedMutex<usize>>, // Store pointer as usize for thread safety
     pub unread_count: std::sync::Arc<std::sync::Mutex<usize>>, // Track messages arrived before callback was set
     pub graph: std::sync::Arc<hiroz::graph::Graph>,
     pub entity: hiroz::entity::EndpointEntity,
@@ -860,11 +885,12 @@ pub extern "C" fn rmw_subscription_get_content_filter(
 
 #[cfg(test)]
 mod gil_deadlock_tests {
+    use std::sync::Arc;
     use std::sync::atomic::{AtomicBool, Ordering};
-    use std::sync::{Arc, Mutex};
     use std::time::{Duration, Instant};
 
     use crate::c_void;
+    use crate::tripwire_compat::GuardedMutex as Mutex;
 
     // Raw C function pointers carry no captured state, so the handshake
     // between the notify thread and the setter thread has to live in
@@ -915,7 +941,7 @@ mod gil_deadlock_tests {
                 gil_wanting_callback as unsafe extern "C" fn(*const std::ffi::c_void, usize),
             )));
         let user_data_holder = Arc::new(Mutex::new(0usize));
-        let unread_count_holder = Arc::new(Mutex::new(0usize));
+        let unread_count_holder = Arc::new(std::sync::Mutex::new(0usize));
 
         let notify = super::build_subscription_notify_callback(
             notifier,
@@ -982,7 +1008,7 @@ mod gil_deadlock_tests {
         let callback_holder: Arc<Mutex<crate::ros::rmw_subscription_new_message_callback_t>> =
             Arc::new(Mutex::new(None));
         let user_data_holder = Arc::new(Mutex::new(0usize));
-        let unread_count_holder = Arc::new(Mutex::new(0usize));
+        let unread_count_holder = Arc::new(std::sync::Mutex::new(0usize));
 
         super::set_subscription_callback_core(
             &callback_holder,

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

@@ -537,9 +537,9 @@ pub extern "C" fn rmw_create_subscription(
 
     // Create shared callback and user_data holders that will be populated after SubscriptionImpl is created
     let callback_holder: std::sync::Arc<
-        std::sync::Mutex<crate::ros::rmw_subscription_new_message_callback_t>,
-    > = std::sync::Arc::new(std::sync::Mutex::new(None));
-    let user_data_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize)); // Store pointer as usize for thread safety
+        crate::tripwire_compat::GuardedMutex<crate::ros::rmw_subscription_new_message_callback_t>,
+    > = std::sync::Arc::new(crate::tripwire_compat::GuardedMutex::new(None));
+    let user_data_holder = std::sync::Arc::new(crate::tripwire_compat::GuardedMutex::new(0usize)); // Store pointer as usize for thread safety
     let unread_count_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize)); // Track unread messages
 
     // Create notification callback that will wake up wait sets and invoke user callback
@@ -1015,9 +1015,9 @@ pub extern "C" fn rmw_create_client(
 
     // Create shared callback and user_data holders
     let callback_holder: std::sync::Arc<
-        std::sync::Mutex<crate::ros::rmw_client_new_response_callback_t>,
-    > = std::sync::Arc::new(std::sync::Mutex::new(None));
-    let user_data_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize));
+        crate::tripwire_compat::GuardedMutex<crate::ros::rmw_client_new_response_callback_t>,
+    > = std::sync::Arc::new(crate::tripwire_compat::GuardedMutex::new(None));
+    let user_data_holder = std::sync::Arc::new(crate::tripwire_compat::GuardedMutex::new(0usize));
 
     // Build the client (notification callback will be set per-request in send_request)
     let zclient = match zclient_builder.build() {
@@ -1211,9 +1211,9 @@ pub extern "C" fn rmw_create_service(
 
     // Create shared callback and user_data holders that will be populated after ServiceImpl is created
     let callback_holder: std::sync::Arc<
-        std::sync::Mutex<crate::ros::rmw_service_new_request_callback_t>,
-    > = std::sync::Arc::new(std::sync::Mutex::new(None));
-    let user_data_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize)); // Store pointer as usize for thread safety
+        crate::tripwire_compat::GuardedMutex<crate::ros::rmw_service_new_request_callback_t>,
+    > = std::sync::Arc::new(crate::tripwire_compat::GuardedMutex::new(None));
+    let user_data_holder = std::sync::Arc::new(crate::tripwire_compat::GuardedMutex::new(0usize)); // Store pointer as usize for thread safety
     let unread_count_holder = std::sync::Arc::new(std::sync::Mutex::new(0usize)); // Track unread requests
 
     // Create notification callback that will wake up wait sets and invoke user callback