fix(node): bound the emitted Arrow IPC stream length at the producer

apis/rust/node/src/node/arrow_utils/ipc_encode.rs

@@ -853,11 +853,14 @@ impl InputDecoder {
     }
 }
 
-/// Reject an IPC payload larger than [`super::MAX_IPC_BYTES`] before decoding.
-/// Defense-in-depth against an oversized peer-controlled zenoh payload, mirroring
-/// the guard in [`decode_arrow_ipc_zero_copy`](super::decode_arrow_ipc_zero_copy)
-/// (the persistent-decoder paths receive the same untrusted bytes).
-fn check_ipc_size(len: usize) -> eyre::Result<()> {
+/// Reject an IPC stream larger than [`super::MAX_IPC_BYTES`].
+///
+/// Used on both sides so the encode and decode limits agree: as a producer-side
+/// guard (so a stream the producer emits is always decodable by every receiver,
+/// rather than silently dropped — see #2586) and as defense-in-depth on decode
+/// against an oversized peer-controlled zenoh payload, mirroring the guard in
+/// [`decode_arrow_ipc_zero_copy`](super::decode_arrow_ipc_zero_copy).
+pub(crate) fn check_ipc_size(len: usize) -> eyre::Result<()> {
     if len > super::MAX_IPC_BYTES {
         bail!(
             "Arrow IPC payload too large: {len} bytes (max {})",
@@ -1462,6 +1465,24 @@ mod tests {
         assert_eq!(decoded.len(), 0);
     }
 
+    /// A `UInt8` payload whose *encoded stream* would exceed `MAX_IPC_BYTES`
+    /// must be rejected at encode time even though `data_len` itself is within
+    /// the limit: the validity bitmap, alignment padding, and message blocks
+    /// make the stream larger, and every receiver bounds the whole stream. The
+    /// producer must fail loudly rather than emit a stream that is silently
+    /// dropped on the zenoh path. Regression for #2586.
+    #[test]
+    fn uint8_ipc_len_rejects_oversized_stream() {
+        // `data_len == MAX_IPC_BYTES` clears a naive `data_len` check, but the
+        // ~1.125x stream overhead pushes the emitted stream over the cap.
+        let err = uint8_ipc_len(super::super::MAX_IPC_BYTES)
+            .expect_err("stream exceeding MAX_IPC_BYTES must be rejected")
+            .to_string();
+        assert!(err.contains("too large"), "unexpected error: {err}");
+        // A comfortably-small payload still encodes to an in-bounds stream.
+        assert!(uint8_ipc_len(1024).unwrap() <= super::super::MAX_IPC_BYTES);
+    }
+
     /// The schema-once receive path must decode a 0-row (empty) batch, not drop
     /// it. A schema-less batch carries no trailing end-of-stream marker, and
     /// arrow's `StreamDecoder` only emits a zero-length body on the poll *after*

apis/rust/node/src/node/mod.rs

@@ -2712,6 +2712,11 @@ impl SampleAllocator {
     pub fn encode_arrow(&self, array: &ArrayData) -> NodeResult<EncodedSample> {
         let sample = match ipc_encode::PreparedIpc::new(array) {
             Some(prepared) => {
+                // Reject a stream larger than every receiver will accept before
+                // emitting it, rather than sending an undecodable payload that
+                // is silently dropped on the zenoh path (#2586).
+                ipc_encode::check_ipc_size(prepared.byte_len())
+                    .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
                 // Prepare once: size the sample from the prepared layout, then
                 // encode into it — avoids rebuilding the layout + IPC headers.
                 let mut sample = self.allocate(prepared.byte_len())?;
@@ -2723,6 +2728,8 @@ impl SampleAllocator {
             None => {
                 let bytes = ipc_encode::encode_ipc_to_vec(array)
                     .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
+                ipc_encode::check_ipc_size(bytes.len())
+                    .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?;
                 let mut sample = self.allocate(bytes.len())?;
                 sample.copy_from_slice(&bytes);
                 sample