perf(py): decode subscriber samples without copying the payload

The Python callback path went through `RawBytesCdrSerdes::deserialize`,
whose `Output` is an owned `RawBytesMessage` carrying no lifetime, so the
whole payload was `to_vec()`d before the closure ran -- one full copy per
message, scaling with payload size, discarded as soon as msgspec had
decoded out of it.

Adds `ZSubBuilder::build_with_sample_callback`, which hands the callback
the `Sample` so it can borrow the payload and decode straight out of the
network buffer, and switches `hiroz-py` to it. Everything else about the
subscriber is unchanged: same encoding validation, same dispatch rules,
same liveliness and graph registration.

Split out of #250. It shared a call site with that PR's re-entrancy fix,
which is proximity, not a reason to review them together.

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/src/pubsub.rs

@@ -867,6 +867,60 @@ where
         })
     }
 
+    /// Build a callback subscriber that receives the whole [`Sample`], undecoded.
+    ///
+    /// [`Self::build_with_callback`] must hand the callback an owned `S::Output`,
+    /// and [`ZDeserializer::Output`] carries no lifetime — so a serdes that only
+    /// forwards bytes (a language binding's identity codec, say) has no way to
+    /// express "borrow the payload", and must copy the entire message before the
+    /// callback has even seen it. That copy scales with payload size and is pure
+    /// waste when the consumer immediately re-reads the bytes into its own
+    /// representation.
+    ///
+    /// This entry point steps around it: the callback gets the `Sample`, so it
+    /// can borrow the payload (`sample.payload().to_bytes()` is a `Cow` that
+    /// borrows whenever the `ZBuf` is contiguous, which the receive path makes it)
+    /// and decode straight out of the network buffer. It can also reach the
+    /// sample's attachment, encoding and timestamp, which the decoded form drops.
+    ///
+    /// Everything else is identical to `build_with_callback` — same encoding
+    /// validation, same [`CallbackDispatcher`] handling, same liveliness and
+    /// graph registration. The callback is user code and is dispatched by exactly
+    /// the same rules.
+    ///
+    /// # Ownership
+    ///
+    /// As with `build_with_callback`, the returned [`ZSub`] must be kept alive for
+    /// the subscription to stay active.
+    pub fn build_with_sample_callback<F>(self, callback: F) -> Result<ZSub<T, (), S>>
+    where
+        F: Fn(Sample) + Send + Sync + 'static,
+        S: ZDeserializer,
+    {
+        let expected_encoding = self.expected_encoding.clone();
+        let callback = Arc::new(move |sample: Sample| {
+            if let Some(ref expected) = expected_encoding {
+                let encoding_str = sample.encoding().to_string();
+                if let Some(received) =
+                    crate::encoding::Encoding::from_zenoh_encoding(&encoding_str)
+                {
+                    if &received != expected {
+                        tracing::warn!(
+                            "Encoding mismatch: expected {:?}, received {:?}",
+                            expected,
+                            received
+                        );
+                    }
+                } else {
+                    tracing::debug!("Unknown encoding format: {}", encoding_str);
+                }
+            }
+            callback(sample);
+        });
+
+        self.build_internal(DataHandler::Callback(callback), None)
+    }
+
     /// Build a subscriber with a callback that processes deserialized messages directly.
     ///
     /// This method creates a subscriber that invokes the provided callback for each