fix(pubsub): stop running subscriber callbacks on the publishing thread

Publishing from inside a subscriber callback on the same session deadlocked
deterministically, with no error and no traceback. Three separate mechanisms
had to be removed.

1. Every subscriber was declared as a zenoh-ext `AdvancedSubscriber`,
   unconditionally. Its sample callback takes a `std::sync::Mutex` and then
   invokes the user callback under that guard. `std::sync::Mutex` is not
   reentrant, so a callback that published into its own topic graph re-entered
   a mutex its own thread already held further up the stack. Declare the
   advanced subscriber only when the QoS profile actually configures advanced
   features: for Volatile — the ROS 2 default — an `AdvancedSubscriber`
   declares no liveliness subscriber, no heartbeat subscriber and no detection
   token, so it was pure per-sample overhead plus the fatal lock.

2. Zenoh delivers a same-session sample synchronously, inline on the thread
   that called `put`. With the lock gone, a callback that publishes therefore
   *recursed* instead of iterating, until the stack overflowed. Adopt the shape
   zenoh's own `FifoChannel` uses, and that zenoh-python installs by default
   for a Python callable: delivery enqueues and returns, user code runs
   elsewhere. `ZPub`'s four publish paths hold a thread-local marker across the
   zenoh `put`; the plain subscriber's shim enqueues when that marker is set
   and invokes inline when it is not. An inter-process sample arrives on a
   zenoh RX worker, which is never inside a hiroz publish, so that path keeps
   its inline call and pays one thread-local read. Queue-mode subscribers —
   every rmw subscription, and every `recv()`-based user — already enqueue and
   return, so they get no dispatcher at all.

   The marker keys on the publishing thread rather than on zenoh's `Locality`
   on purpose: two sessions in one process with a direct route deliver a
   `Locality::Remote` sample inline on the publisher's thread, so an
   `allowed_origin(SessionLocal)` split would miss that case.

   With no path left on which a callback is reachable from inside `put`,
   re-entrancy is structurally impossible rather than depth-bounded, so the
   interim `MAX_CALLBACK_REENTRY_DEPTH` cap, `CallbackDepthGuard` and
   `InheritedCallbackDepth` are removed rather than left as unreachable
   defensive code.

3. `hiroz-py`'s `ZPublisher.publish`/`publish_raw` release the GIL across the
   zenoh publish. This does not fix the deadlock, but it downgrades a
   whole-interpreter freeze — no exception, no traceback, only an external
   kill — to a single blocked thread, which is the difference between an
   undiagnosable hang and a diagnosable one.

Queue policy is split per path. The plain dispatcher takes its capacity from
the same history-QoS expression `build()` uses to size `BoundedQueue` and drops
the oldest on overflow; the advanced dispatcher stays unbounded. Bounded and
blocking recreates the deadlock on both paths — the blocked producer sits
inside the callback holding the very lock the drain thread needs. Bounded and
dropping is correct for the plain path, which is Volatile with KEEP_LAST(depth)
and already promises no more than `depth` undelivered samples, and wrong for the
advanced path, where loss would discard samples miss-detection went out of its
way to recover, mid-reorder.

Also adds `ZSubBuilder::build_with_sample_callback`, which hands the callback
the `Sample` rather than a decoded message, and uses it in `hiroz-py`. The
Python callback path routed every sample through an identity codec whose
`Output` carries no lifetime, so it had to `to_vec()` the whole payload before
the callback ran, only for msgspec to decode out of it and drop it — one full
payload copy per message. Measured as an interleaved paired A/B (two release
wheels differing only in this call site, 5+6 reps, 16k timed round trips each,
half-trip p50): 64 B 114.5 -> 114.1 (noise), 4 KB 120.9 -> 121.0 (noise),
64 KB 224.4 -> 221.1, -3.4 us with 10 of 11 reps in [-2.6, -5.1]. The
size-dependence is the point: it is what distinguishes removing a
payload-sized memcpy from removing a fixed per-message cost.

Wire format is unchanged: an `AdvancedPublisher` with no cache, no
publisher_detection and no sample_miss_detection puts on the plain key
expression, so Volatile publishers and subscribers stay byte-identical and
interop is unaffected. Volatile subscribers no longer get zenoh-ext's
HLC-timestamp de-duplication.

Ordering is preserved where it was ever guaranteed: one FIFO queue, one drain
thread, and drop-oldest preserves the relative order of what survives. Not
preserved, and documented at the type: a plain subscriber receiving both local
and remote publications on one topic now runs them on two different threads, so
their relative order is not guaranteed. Neither ROS 2 nor zenoh guarantees
ordering across publishers, and a plain zenoh subscriber could already be
invoked concurrently from several RX workers.

Detector evidence, both directions. Reverting the dispatch decision
(`local_publish_active() -> false`) and keeping the tests, all three
self-feeding loop tests die with `fatal runtime error: stack overflow,
aborting` (rc=101), including `intra_closed_loop_runs_iteratively`, which
drives 2_000 round trips through a two-topic one-session callback cycle that
could not previously be expressed as a loop. Against the unfixed sources the
original three scenarios fail on their 20s deadline (`0 passed; 3 failed`).
Every one of the six Python cells hangs with rc=124 under an external
wall-clock timeout on a wheel built from unfixed sources — not merely fails,
hangs — and all six pass on this branch.

Author: YuanYuYuan

crates/hiroz-py/src/node.rs

@@ -16,6 +16,7 @@ use hiroz::node::ZNode;
 use pyo3::prelude::*;
 use std::any::Any;
 use std::sync::Arc;
+use zenoh_buffers::buffer::SplitBuffer;
 
 /// Try to extract type info from a message class.
 ///
@@ -233,9 +234,25 @@ impl PyZNode {
             // matching rmw_zenoh_cpp's NodeData::subs_ pattern. The caller does not
             // need to assign the returned PyZSubscriber to keep the subscription active.
             let type_name = msg_type_str.clone();
+            // Sample-level callback, not `build_with_callback`. The typed form
+            // would route through `RawBytesCdrSerdes::deserialize`, whose
+            // `Output` is an owned `RawBytesMessage` and so must `to_vec()` the
+            // whole payload before this closure runs — a full copy per message,
+            // scaling with payload size, immediately discarded once msgspec has
+            // decoded it. Taking the `Sample` lets the decode read straight out
+            // of the network buffer, and matches what the polling `recv()` path
+            // in `pubsub.rs` already does.
             let zsub = sub_builder
-                .build_with_callback(move |raw_msg: RawBytesMessage| {
-                    let payload = raw_msg.0;
+                .build_with_sample_callback(move |sample| {
+                    // Same zero-copy setup as `PyZSubscriber::recv`: the ZBuf is
+                    // cheap Arc clones, and publishing it as the deserializer's
+                    // source lets `bytes`-typed fields become sub-ZSlices of the
+                    // received buffer instead of copies.
+                    let payload_zbuf: zenoh_buffers::ZBuf = sample.payload().clone().into();
+                    hiroz_cdr::ZBUF_DESER_SOURCE.with(|cell| {
+                        *cell.borrow_mut() = Some(payload_zbuf.clone());
+                    });
+                    let payload = payload_zbuf.contiguous();
                     Python::with_gil(|py| {
                         match hiroz_msgs::deserialize_from_cdr(&type_name, py, &payload) {
                             Ok(obj) => {
@@ -248,6 +265,9 @@ impl PyZNode {
                             }
                         }
                     });
+                    hiroz_cdr::ZBUF_DESER_SOURCE.with(|cell| {
+                        *cell.borrow_mut() = None;
+                    });
                 })
                 .map_err(|e| e.into_pyerr())?;
 

crates/hiroz-py/src/pubsub.rs

@@ -24,20 +24,30 @@ impl PyZPublisher {
     /// Publish a message
     ///
     /// Serializes the Python message (msgspec.Struct) to ZBuf and publishes (zero-copy path)
-    unsafe fn publish(&self, _py: Python, data: &Bound<'_, PyAny>) -> PyResult<()> {
-        // Serialize Python message directly to ZBuf (zero-copy)
+    unsafe fn publish(&self, py: Python, data: &Bound<'_, PyAny>) -> PyResult<()> {
+        // Serialize Python message directly to ZBuf (zero-copy). This touches
+        // Python objects, so it must run with the GIL held.
         let zbuf = hiroz_msgs::serialize_to_zbuf(&self.type_name, data)?;
 
-        // Publish the ZBuf directly
-        self.inner.publish(zbuf.into()).map_err(|e| e.into_pyerr())
+        // Release the GIL for the publish itself. Zenoh delivers samples to
+        // local subscribers synchronously on the publishing thread, so a publish
+        // issued from inside a subscriber callback can block here; holding the
+        // GIL across it would freeze the whole interpreter — no exception, no
+        // traceback — instead of blocking just this thread.
+        py.allow_threads(|| self.inner.publish(zbuf.into()))
+            .map_err(|e| e.into_pyerr())
     }
 
     /// Publish pre-serialized CDR bytes directly
     ///
     /// Use this for zero-copy forwarding of received messages (e.g., in a pong responder).
     /// The bytes should be in CDR format (as returned by recv_serialized/try_recv_serialized).
-    fn publish_raw(&self, data: &[u8]) -> PyResult<()> {
-        self.inner.publish(data.into()).map_err(|e| e.into_pyerr())
+    fn publish_raw(&self, py: Python, data: &[u8]) -> PyResult<()> {
+        // Copy out of the Python buffer before dropping the GIL, then publish
+        // without it — see `publish` for why.
+        let payload: zenoh::bytes::ZBytes = data.into();
+        py.allow_threads(|| self.inner.publish(payload))
+            .map_err(|e| e.into_pyerr())
     }
 
     /// Get the topic name (for debugging)