fix(parameter): stop deadlocking on re-entry from an on_set callback

`ParameterState::validate_and_apply` bound the `on_set_callback.read()` guard
to a named local and invoked the user callback under it. Two re-entrant paths,
both reachable from the public API:

- `on_set_parameters` from inside the callback takes `on_set_callback.write()`
  while the same thread still holds the read guard — a guaranteed self-deadlock,
  no race required.
- `set_parameter` from inside the callback re-enters `validate_and_apply` and
  takes the read lock recursively, which `std::sync::RwLock` does not
  guarantee: it deadlocks if a writer is queued between the two acquisitions.

`SetCallback` is already an `Arc`, so the fix is to clone it out and let the
guard drop before the call — one refcount bump, no structural change. The store
guard in the same function was already correctly scoped; only the callback
guard was wrong.

The lock becomes a `TrackedRwLock` and both call sites dispatch through
`invoke_user_callback!`, so a reintroduction panics in debug naming the site
instead of hanging.

Detector evidence, both directions. Against the unfixed source
`parameter_on_set_callback_reregistering_does_not_deadlock` fails on its 30s
deadline (3 passed; 1 failed, 35.59s); with the fix all four pass in 6.59s.
Separately, reverting only the guard-drop and keeping the tripwire makes the
violation fire on `test_parameter_validation_callback` — an *ordinary* test,
not a deadlock test — naming the site and the live guard count.

Services and actions were audited for the same shape and are clean.
`ZServer::build_internal` declares a plain zenoh queryable and calls
`handler.handle(query)` with no hiroz lock held; the action server's user
handler is awaited with no guard on the stack, and the action client never runs
user code on a zenoh thread at all.

Author: YuanYuYuan

crates/hiroz-tests/tests/reentrant_service.rs

@@ -0,0 +1,348 @@
+//! Re-entrancy audit for the subsystems the pub/sub deadlock fix did *not* touch.
+//!
+//! The pub/sub bug was specific: zenoh-ext's `AdvancedSubscriber::sub_callback`
+//! takes a non-reentrant `std::sync::Mutex` and invokes the user callback under
+//! that guard, so a callback that published back into its own session re-entered
+//! a mutex its own thread already held.
+//!
+//! Services, actions and parameters use plain zenoh (`declare_queryable`), and
+//! zenoh core deliberately clones the queryable callbacks and `drop(state)`s
+//! before invoking them (`zenoh/src/api/session.rs`, `handle_query`) — exactly as
+//! it does for subscribers in `resolve_put`. That makes them *likely* safe, but
+//! "likely, by analogy" is not evidence. These tests are the evidence.
+//!
+//! Every scenario runs on a dedicated thread behind a hard deadline, so a
+//! re-entrancy deadlock fails the test instead of wedging the suite — the same
+//! shape as `reentrant_publish.rs`.
+//!
+//! Note on which API each test exercises. hiroz's *ergonomic* service server
+//! (`create_service(..).build()`) is queue-mode: the query is pushed onto a
+//! `BoundedQueue` and the user drains it with `take_request()` from their own
+//! thread, so ordinary service handlers never run on a zenoh RX thread at all.
+//! `build_with_callback` is the raw escape hatch that does put user code on the
+//! RX thread (it is what the parameter service and the action server use
+//! internally), so that is the path these tests target — it is the only place a
+//! service-side re-entrancy deadlock could exist.
+
+#![cfg(feature = "ros-msgs")]
+
+mod common;
+
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicBool, AtomicUsize, Ordering},
+        mpsc,
+    },
+    thread,
+    time::Duration,
+};
+
+use common::{TestRouter, create_hiroz_context_with_endpoint};
+use hiroz::{
+    Builder,
+    msg::{SerdeCdrSerdes, ZSerializer},
+    parameter::{Parameter, ParameterValue, SetParametersResult},
+};
+use hiroz_msgs::example_interfaces::{AddTwoIntsRequest, AddTwoIntsResponse, srv::AddTwoInts};
+use serial_test::serial;
+use zenoh::{Wait, query::Query};
+
+/// Budget for one scenario. Generous relative to the work done — anything slower
+/// than this is a hang, not slowness.
+const SCENARIO_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Run `scenario` on its own thread; fail (rather than hang) past the deadline.
+///
+/// On timeout the worker is deliberately left running: it is blocked on a lock
+/// that will never be released, and there is no sound way to unwind it.
+fn with_deadline(name: &'static str, scenario: impl FnOnce() + Send + 'static) {
+    let (tx, rx) = mpsc::channel();
+    thread::spawn(move || {
+        scenario();
+        let _ = tx.send(());
+    });
+    match rx.recv_timeout(SCENARIO_TIMEOUT) {
+        Ok(()) => {}
+        // The worker panicked and dropped the sender. That is an assertion
+        // failure inside the scenario, NOT a deadlock — reporting it as one
+        // would turn every ordinary test failure into a false deadlock report.
+        Err(mpsc::RecvTimeoutError::Disconnected) => {
+            panic!("{name}: scenario panicked — see the worker thread's panic above")
+        }
+        Err(mpsc::RecvTimeoutError::Timeout) => {
+            panic!("{name}: scenario did not finish within {SCENARIO_TIMEOUT:?} — deadlock")
+        }
+    }
+}
+
+/// Reply to a query, echoing the attachment so the hiroz client can match it.
+fn reply_sum(query: &Query, sum: i64) {
+    let bytes = SerdeCdrSerdes::<AddTwoIntsResponse>::serialize(&AddTwoIntsResponse { sum });
+    let mut reply = query.reply(query.key_expr().clone(), bytes);
+    if let Some(att) = query.attachment() {
+        reply = reply.attachment(att.clone());
+    }
+    let _ = reply.wait();
+}
+
+/// A service handler that publishes on the same session.
+///
+/// The handler runs on the zenoh RX thread, inside the queryable callback. If
+/// hiroz held any lock across `handler.handle(query)` — as the pub/sub path used
+/// to — this publish would re-enter it.
+#[test]
+#[serial]
+fn service_handler_publishing_does_not_deadlock() {
+    with_deadline("service_handler_publishing", || {
+        let router = TestRouter::new();
+        let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("ctx");
+        let node = ctx.create_node("svc_pub").build().expect("node");
+
+        let published = Arc::new(AtomicUsize::new(0));
+        let seen = Arc::new(AtomicUsize::new(0));
+
+        let sub_seen = seen.clone();
+        let _sub = node
+            .create_sub::<hiroz_msgs::std_msgs::String>("/svc_side_effect")
+            .build_with_callback(move |_m| {
+                sub_seen.fetch_add(1, Ordering::SeqCst);
+            })
+            .expect("sub");
+
+        let side_pub = Arc::new(
+            node.create_pub::<hiroz_msgs::std_msgs::String>("/svc_side_effect")
+                .build()
+                .expect("pub"),
+        );
+
+        let pub_for_handler = side_pub.clone();
+        let published_c = published.clone();
+        let _server = node
+            .create_service::<AddTwoInts>("add_two_ints")
+            .build_with_callback(move |query: Query| {
+                // The re-entrant side effect: publish from inside the queryable
+                // callback, on the same session.
+                pub_for_handler
+                    .publish(&hiroz_msgs::std_msgs::String {
+                        data: "from-service-handler".into(),
+                    })
+                    .expect("publish from service handler");
+                published_c.fetch_add(1, Ordering::SeqCst);
+                reply_sum(&query, 42);
+            })
+            .expect("server");
+
+        let client = node
+            .create_client::<AddTwoInts>("add_two_ints")
+            .build()
+            .expect("client");
+
+        thread::sleep(Duration::from_millis(1000));
+
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let resp = rt.block_on(async {
+            client
+                .call_with_timeout(&AddTwoIntsRequest { a: 1, b: 2 }, Duration::from_secs(10))
+                .await
+        });
+        assert!(resp.is_ok(), "service call failed: {:?}", resp.err());
+
+        thread::sleep(Duration::from_millis(500));
+        assert_eq!(
+            published.load(Ordering::SeqCst),
+            1,
+            "handler did not complete its publish"
+        );
+        assert!(
+            seen.load(Ordering::SeqCst) >= 1,
+            "the publish issued from the service handler was never delivered"
+        );
+    });
+}
+
+/// A service handler that calls a *second* service on the same session.
+///
+/// The nested call is issued from the zenoh RX thread. This is the "service
+/// handler calls another service" hazard: if the queryable dispatch path held a
+/// lock, or if the inner query could only be answered by the very thread that is
+/// blocked, this never returns.
+#[test]
+#[serial]
+fn service_handler_calling_another_service_does_not_deadlock() {
+    with_deadline("service_handler_nested_call", || {
+        let router = TestRouter::new();
+        let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("ctx");
+        let node = ctx.create_node("svc_nested").build().expect("node");
+
+        let _inner = node
+            .create_service::<AddTwoInts>("inner")
+            .build_with_callback(move |query: Query| reply_sum(&query, 7))
+            .expect("inner server");
+
+        let inner_client = Arc::new(
+            node.create_client::<AddTwoInts>("inner")
+                .build()
+                .expect("inner client"),
+        );
+
+        let nested_ok = Arc::new(AtomicBool::new(false));
+        let nested_ok_c = nested_ok.clone();
+        let inner_for_handler = inner_client.clone();
+
+        let _outer = node
+            .create_service::<AddTwoInts>("outer")
+            .build_with_callback(move |query: Query| {
+                // Nested service call from inside a queryable callback.
+                //
+                // The callback runs on one of zenoh's own tokio worker threads,
+                // so `Runtime::block_on` here panics with "Cannot start a
+                // runtime from within a runtime". The nested call therefore runs
+                // on a plain std thread with its own runtime, and this callback
+                // *joins* it — which is the hazard being tested: the zenoh RX
+                // thread is blocked for the whole duration of the inner query.
+                // If serving that inner query required this very thread, the
+                // join never returns.
+                let client = inner_for_handler.clone();
+                let worker = thread::spawn(move || {
+                    let rt = tokio::runtime::Builder::new_current_thread()
+                        .enable_all()
+                        .build()
+                        .unwrap();
+                    rt.block_on(async {
+                        client
+                            .call_with_timeout(
+                                &AddTwoIntsRequest { a: 3, b: 4 },
+                                Duration::from_secs(10),
+                            )
+                            .await
+                    })
+                });
+                let inner = worker.join().expect("nested-call worker panicked");
+                if inner.is_ok() {
+                    nested_ok_c.store(true, Ordering::SeqCst);
+                }
+                reply_sum(&query, inner.map(|r| r.sum).unwrap_or(-1));
+            })
+            .expect("outer server");
+
+        let outer_client = node
+            .create_client::<AddTwoInts>("outer")
+            .build()
+            .expect("outer client");
+
+        thread::sleep(Duration::from_millis(1000));
+
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let resp = rt.block_on(async {
+            outer_client
+                .call_with_timeout(&AddTwoIntsRequest { a: 1, b: 2 }, Duration::from_secs(20))
+                .await
+        });
+        assert!(resp.is_ok(), "outer service call failed: {:?}", resp.err());
+        assert!(
+            nested_ok.load(Ordering::SeqCst),
+            "the nested service call from inside the handler did not complete"
+        );
+        assert_eq!(resp.unwrap().sum, 7, "nested result not propagated");
+    });
+}
+
+/// A parameter `on_set` callback that sets another parameter.
+///
+/// `ParameterState::validate_and_apply` invokes the user callback while holding
+/// `on_set_callback.read()` (an `std::sync::RwLock` read guard). A callback that
+/// calls `set_parameter` re-enters `validate_and_apply` on the same thread and
+/// therefore takes that same read lock recursively. Recursive read acquisition on
+/// `std::sync::RwLock` is explicitly not guaranteed by the standard library — it
+/// deadlocks if a writer is queued between the two acquisitions — so this is the
+/// closest analogue to the pub/sub bug outside pub/sub.
+#[test]
+#[serial]
+fn parameter_on_set_callback_setting_another_parameter_does_not_deadlock() {
+    with_deadline("parameter_on_set_reentrant", || {
+        let router = TestRouter::new();
+        let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("ctx");
+        let node = Arc::new(ctx.create_node("param_reentrant").build().expect("node"));
+
+        node.declare_parameter("a", ParameterValue::Integer(0), Default::default())
+            .expect("declare a");
+        node.declare_parameter("b", ParameterValue::Integer(0), Default::default())
+            .expect("declare b");
+
+        let reentered = Arc::new(AtomicBool::new(false));
+        let reentered_c = reentered.clone();
+        // Weak, so the callback (owned by the node) does not keep the node alive.
+        let node_weak = Arc::downgrade(&node);
+
+        node.on_set_parameters(move |changed: &[Parameter]| {
+            // Only re-enter for "a", or this recurses forever.
+            if changed.iter().any(|p| p.name == "a")
+                && !reentered_c.swap(true, Ordering::SeqCst)
+                && let Some(n) = node_weak.upgrade()
+            {
+                // Re-entrant set from inside the on_set callback.
+                let _ = n.set_parameter(Parameter::new("b", ParameterValue::Integer(99)));
+            }
+            SetParametersResult::success()
+        });
+
+        node.set_parameter(Parameter::new("a", ParameterValue::Integer(1)))
+            .expect("set a");
+
+        assert!(
+            reentered.load(Ordering::SeqCst),
+            "the on_set callback never ran"
+        );
+        assert_eq!(
+            node.get_parameter("b"),
+            Some(ParameterValue::Integer(99)),
+            "the re-entrant set_parameter did not take effect"
+        );
+    });
+}
+
+/// A parameter `on_set` callback that replaces the callback registration.
+///
+/// This is the deterministic form of the same defect. `validate_and_apply` holds
+/// `on_set_callback.read()` across the user callback; `on_set_parameters` takes
+/// `on_set_callback.write()`. A callback that re-registers therefore asks the
+/// same thread for a write lock while it still holds a read lock on the same
+/// `std::sync::RwLock` — a guaranteed self-deadlock, no race required.
+///
+/// "Swap out the validator once the node is configured" is an ordinary thing to
+/// want, and rclcpp supports it (`remove_on_set_parameters_callback` /
+/// `add_on_set_parameters_callback` are callable from within a callback), so
+/// this is a reachable API shape rather than a contrived one.
+#[test]
+#[serial]
+fn parameter_on_set_callback_reregistering_does_not_deadlock() {
+    with_deadline("parameter_on_set_reregister", || {
+        let router = TestRouter::new();
+        let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("ctx");
+        let node = Arc::new(ctx.create_node("param_rereg").build().expect("node"));
+
+        node.declare_parameter("a", ParameterValue::Integer(0), Default::default())
+            .expect("declare a");
+
+        let ran = Arc::new(AtomicBool::new(false));
+        let ran_c = ran.clone();
+        let node_weak = Arc::downgrade(&node);
+
+        node.on_set_parameters(move |_changed: &[Parameter]| {
+            if !ran_c.swap(true, Ordering::SeqCst)
+                && let Some(n) = node_weak.upgrade()
+            {
+                // Re-register from inside the callback: write lock requested
+                // while this thread still holds the read lock.
+                n.on_set_parameters(|_| SetParametersResult::success());
+            }
+            SetParametersResult::success()
+        });
+
+        node.set_parameter(Parameter::new("a", ParameterValue::Integer(1)))
+            .expect("set a");
+
+        assert!(ran.load(Ordering::SeqCst), "the on_set callback never ran");
+    });
+}