corpus_ipc/zmq_message/
transport.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use std::sync::{Mutex, MutexGuard};
4
5use crate::validation::Validate;
6use crate::zmq_context::shared_context;
7use crate::{
8    CanonicalEncodeError, IpcMessage, ValidationKind, decode_ipc_message_json,
9    encode_canonical_ipc_message,
10};
11
12/// Why a ZeroMQ message was not a single structured frame.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
14pub enum FramingFault {
15    /// The frame contained no bytes.
16    #[error("message was empty")]
17    Empty,
18    /// libzmq delivered more than one frame for the message.
19    #[error("message had multiple ZeroMQ frames; exactly one JSON envelope frame is required")]
20    Multipart,
21}
22
23/// Failures from the structured ZeroMQ publisher and subscriber.
24///
25/// Outcomes are matchable. Library paths return one of these variants instead
26/// of panicking on malformed input, version mismatch, high-water-mark pressure,
27/// or a closed socket.
28#[derive(Debug, thiserror::Error)]
29pub enum ZmqMessageError {
30    /// [`ZmqTransportConfig`] rejected the caller's settings before any socket call.
31    #[error("invalid transport config: {0}")]
32    InvalidConfig(&'static str),
33
34    /// Socket creation, option setup, bind, or connect failed.
35    #[error("ZeroMQ socket setup failed: {0}")]
36    Setup(String),
37
38    /// The message failed [`Validate`] before encoding.
39    #[error("payload failed validation: {0}")]
40    Validation(#[from] crate::ValidationError),
41
42    /// Canonical encoding failed after validation passed.
43    #[error("canonical encoding failed: {0}")]
44    Encode(CanonicalEncodeError),
45
46    /// The encoded frame is larger than [`STRUCTURED_MAX_FRAME_BYTES`].
47    #[error("structured frame is too large: {actual} bytes (maximum {max} bytes)")]
48    FrameTooLarge {
49        /// Encoded length in bytes.
50        actual: usize,
51        /// [`STRUCTURED_MAX_FRAME_BYTES`] as `usize`.
52        max: usize,
53    },
54
55    /// The frame was not an acceptable [`crate::WireEnvelope`] or legacy payload.
56    #[error("envelope decode failed: {0}")]
57    Envelope(#[from] crate::EnvelopeError),
58
59    /// The ZeroMQ message was not exactly one non-empty frame.
60    #[error("unexpected ZeroMQ framing: {0}")]
61    UnexpectedFraming(FramingFault),
62
63    /// Non-blocking receive found no message (`EAGAIN`).
64    #[error("no message available (EAGAIN)")]
65    WouldBlock,
66
67    /// Non-blocking send returned `EAGAIN` because the outbound queue is full.
68    ///
69    /// PUB sockets more often drop silently when the high-water mark is reached
70    /// and `send` still returns `Ok`. This variant is the case where libzmq
71    /// reports the pressure instead of dropping.
72    #[error("outbound high-water mark is full (EAGAIN on send)")]
73    HighWaterMark,
74
75    /// `close` already dropped the socket, or it was never stored.
76    #[error("ZeroMQ socket is closed")]
77    Closed,
78
79    /// A libzmq send or receive error other than `EAGAIN`.
80    #[error("ZeroMQ transport error: {0}")]
81    Transport(String),
82}
83
84/// Endpoint and socket options for one structured PUB or SUB socket.
85///
86/// Options are applied before `bind` or `connect`. A high-water mark of `0`
87/// is unlimited: libzmq does not treat `0` as an empty queue. Linger `-1`
88/// means wait forever on close. Inbound frames are capped at
89/// [`STRUCTURED_MAX_FRAME_BYTES`].
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct ZmqTransportConfig {
92    /// ZeroMQ endpoint (`ipc://…` or `tcp://…`). The publisher binds it; the
93    /// subscriber connects to it.
94    pub endpoint: String,
95    /// `ZMQ_SNDHWM`. Must be `>= 0`. `0` means unlimited.
96    pub send_hwm: i32,
97    /// `ZMQ_RCVHWM`. Must be `>= 0`. `0` means unlimited.
98    pub receive_hwm: i32,
99    /// `ZMQ_LINGER` in milliseconds. `-1` waits forever; `0` discards unsent
100    /// frames on close. Must be `>= -1`.
101    pub linger_ms: i32,
102    /// `ZMQ_RECONNECT_IVL` in milliseconds. Must be `> 0`.
103    pub reconnect_ivl_ms: i32,
104    /// `ZMQ_RECONNECT_IVL_MAX` in milliseconds. Must be `> 0` and at least
105    /// [`Self::reconnect_ivl_ms`]. libzmq ignores a cap below the interval.
106    pub reconnect_ivl_max_ms: i32,
107}
108
109impl ZmqTransportConfig {
110    /// Config with libzmq's default high-water marks (1000), a 100 ms reconnect
111    /// interval, a 5 s reconnect cap, and linger `0`.
112    ///
113    /// Linger `0` keeps teardown non-blocking. Undelivered frames are discarded,
114    /// which matches at-most-once delivery.
115    ///
116    /// ```
117    /// use corpus_ipc::ZmqTransportConfig;
118    ///
119    /// let config = ZmqTransportConfig::new("tcp://127.0.0.1:9");
120    /// assert_eq!(config.linger_ms, 0);
121    /// assert_eq!(config.send_hwm, 1000);
122    /// assert_eq!(config.reconnect_ivl_max_ms, 5_000);
123    /// ```
124    #[must_use]
125    pub fn new(endpoint: impl Into<String>) -> Self {
126        Self {
127            endpoint: endpoint.into(),
128            send_hwm: 1000,
129            receive_hwm: 1000,
130            linger_ms: 0,
131            reconnect_ivl_ms: 100,
132            reconnect_ivl_max_ms: 5_000,
133        }
134    }
135
136    fn check(&self) -> Result<(), ZmqMessageError> {
137        self.endpoint_failure()?;
138        self.limit_failure()?;
139        self.reconnect_failure()
140    }
141
142    fn endpoint_failure(&self) -> Result<(), ZmqMessageError> {
143        if self.endpoint.is_empty() {
144            return Err(invalid_config("endpoint must be non-empty"));
145        }
146        if self.endpoint.as_bytes().contains(&0) {
147            return Err(invalid_config("endpoint must not contain an interior null"));
148        }
149        Ok(())
150    }
151
152    fn limit_failure(&self) -> Result<(), ZmqMessageError> {
153        if self.send_hwm < 0 {
154            return Err(invalid_config("send_hwm must be >= 0"));
155        }
156        if self.receive_hwm < 0 {
157            return Err(invalid_config("receive_hwm must be >= 0"));
158        }
159        if self.linger_ms < -1 {
160            return Err(invalid_config("linger_ms must be >= -1"));
161        }
162        Ok(())
163    }
164
165    fn reconnect_failure(&self) -> Result<(), ZmqMessageError> {
166        if self.reconnect_ivl_ms <= 0 {
167            return Err(invalid_config("reconnect_ivl_ms must be > 0"));
168        }
169        if self.reconnect_ivl_max_ms <= 0 {
170            return Err(invalid_config("reconnect_ivl_max_ms must be > 0"));
171        }
172        if self.reconnect_ivl_max_ms < self.reconnect_ivl_ms {
173            return Err(invalid_config(
174                "reconnect_ivl_max_ms must be >= reconnect_ivl_ms",
175            ));
176        }
177        Ok(())
178    }
179}
180
181fn invalid_config(message: &'static str) -> ZmqMessageError {
182    ZmqMessageError::InvalidConfig(message)
183}
184
185pub(super) fn ensure_frame_fits(len: usize) -> Result<(), ZmqMessageError> {
186    let max = usize::try_from(STRUCTURED_MAX_FRAME_BYTES).unwrap_or(usize::MAX);
187    if len > max {
188        return Err(ZmqMessageError::FrameTooLarge { actual: len, max });
189    }
190    Ok(())
191}
192
193/// Largest structured frame a publisher or subscriber will accept (`ZMQ_MAXMSGSIZE`).
194///
195/// 512 MiB covers a [`crate::ProtocolLimits::DEFAULT`] gradient batch of 65_536
196/// layer ids at the 1_024-byte string cap when every byte is JSON-escaped
197/// (`\u00XX`), plus row framing and the aggregate float budget. [`ZmqMessagePublisher::send`]
198/// rejects a larger encoded frame before it hits the socket. libzmq applies the
199/// same cap on receive, before [`zmq::Socket::recv_bytes`] allocates the frame.
200pub const STRUCTURED_MAX_FRAME_BYTES: i64 = 512 * 1024 * 1024;
201
202/// Exclusive owner of a libzmq socket.
203///
204/// Inherits `Send` + `!Sync` from [`zmq::Socket`]. Callers reach it only
205/// through the backend mutex, and only with `&mut self`.
206struct ExclusiveSocket {
207    socket: zmq::Socket,
208}
209
210impl ExclusiveSocket {
211    fn send(&mut self, bytes: &[u8], flags: i32) -> zmq::Result<()> {
212        self.socket.send(bytes, flags)
213    }
214
215    fn recv_bytes(&mut self, flags: i32) -> zmq::Result<Vec<u8>> {
216        self.socket.recv_bytes(flags)
217    }
218
219    fn rcvmore(&mut self) -> zmq::Result<bool> {
220        self.socket.get_rcvmore()
221    }
222
223    fn option_snapshot(&self) -> zmq::Result<(i32, i32, i32, i32, i32, i64)> {
224        Ok((
225            self.socket.get_sndhwm()?,
226            self.socket.get_rcvhwm()?,
227            self.socket.get_linger()?,
228            self.socket.get_reconnect_ivl()?,
229            self.socket.get_reconnect_ivl_max()?,
230            self.socket.get_maxmsgsize()?,
231        ))
232    }
233}
234
235enum SocketMode {
236    Bind,
237    Connect,
238}
239
240fn apply_options(socket: &zmq::Socket, config: &ZmqTransportConfig) -> Result<(), ZmqMessageError> {
241    socket
242        .set_sndhwm(config.send_hwm)
243        .map_err(|err| ZmqMessageError::Setup(format!("sndhwm: {err}")))?;
244    socket
245        .set_rcvhwm(config.receive_hwm)
246        .map_err(|err| ZmqMessageError::Setup(format!("rcvhwm: {err}")))?;
247    socket
248        .set_linger(config.linger_ms)
249        .map_err(|err| ZmqMessageError::Setup(format!("linger: {err}")))?;
250    socket
251        .set_reconnect_ivl(config.reconnect_ivl_ms)
252        .map_err(|err| ZmqMessageError::Setup(format!("reconnect_ivl: {err}")))?;
253    socket
254        .set_reconnect_ivl_max(config.reconnect_ivl_max_ms)
255        .map_err(|err| ZmqMessageError::Setup(format!("reconnect_ivl_max: {err}")))?;
256    socket
257        .set_maxmsgsize(STRUCTURED_MAX_FRAME_BYTES)
258        .map_err(|err| ZmqMessageError::Setup(format!("maxmsgsize: {err}")))?;
259    Ok(())
260}
261
262fn open_socket(
263    context: &zmq::Context,
264    kind: zmq::SocketType,
265    config: &ZmqTransportConfig,
266    mode: SocketMode,
267) -> Result<ExclusiveSocket, ZmqMessageError> {
268    config.check()?;
269    let socket = context
270        .socket(kind)
271        .map_err(|err| ZmqMessageError::Setup(format!("socket: {err}")))?;
272    // Reconnect and queue options must be set before bind/connect. libzmq
273    // reads them when it establishes the socket, not on a later setsockopt.
274    apply_options(&socket, config)?;
275    attach_socket(&socket, config, mode)?;
276    Ok(ExclusiveSocket { socket })
277}
278
279fn attach_socket(
280    socket: &zmq::Socket,
281    config: &ZmqTransportConfig,
282    mode: SocketMode,
283) -> Result<(), ZmqMessageError> {
284    match mode {
285        SocketMode::Bind => bind_socket(socket, config),
286        SocketMode::Connect => connect_socket(socket, config),
287    }
288}
289
290fn bind_socket(socket: &zmq::Socket, config: &ZmqTransportConfig) -> Result<(), ZmqMessageError> {
291    socket
292        .bind(&config.endpoint)
293        .map_err(|err| ZmqMessageError::Setup(format!("bind {}: {err}", config.endpoint)))
294}
295
296fn connect_socket(
297    socket: &zmq::Socket,
298    config: &ZmqTransportConfig,
299) -> Result<(), ZmqMessageError> {
300    socket
301        .set_subscribe(b"")
302        .map_err(|err| ZmqMessageError::Setup(format!("subscribe: {err}")))?;
303    socket
304        .connect(&config.endpoint)
305        .map_err(|err| ZmqMessageError::Setup(format!("connect {}: {err}", config.endpoint)))
306}
307
308fn lock_socket(slot: &Mutex<Option<ExclusiveSocket>>) -> MutexGuard<'_, Option<ExclusiveSocket>> {
309    slot.lock()
310        .unwrap_or_else(std::sync::PoisonError::into_inner)
311}
312
313fn map_send(result: zmq::Result<()>) -> Result<(), ZmqMessageError> {
314    match result {
315        Ok(()) => Ok(()),
316        Err(zmq::Error::EAGAIN) => Err(ZmqMessageError::HighWaterMark),
317        Err(err) => Err(ZmqMessageError::Transport(err.to_string())),
318    }
319}
320
321/// Decode one received frame. Shared by [`ZmqMessageSubscriber::try_recv`]
322/// and [`ZmqMessageSubscriber::decode_frame_for_tests`].
323fn decode_frame(bytes: &[u8]) -> Result<IpcMessage, ZmqMessageError> {
324    if bytes.is_empty() {
325        return Err(ZmqMessageError::UnexpectedFraming(FramingFault::Empty));
326    }
327    let message = decode_ipc_message_json(bytes)?;
328    message.validate()?;
329    Ok(message)
330}
331
332impl std::fmt::Debug for ZmqMessagePublisher {
333    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
334        formatter.write_str("ZmqMessagePublisher")
335    }
336}
337
338/// Publishes one canonical JSON [`IpcMessage`] per ZeroMQ message.
339///
340/// The socket is a `PUB` socket that binds [`ZmqTransportConfig::endpoint`].
341/// `send` validates the payload, then calls [`encode_canonical_ipc_message`],
342/// then writes a single frame with `ZMQ_DONTWAIT`.
343///
344/// See the [module docs](crate::zmq_message) for slow-joiner loss, high-water-mark loss, and
345/// linger. `Ok` from [`Self::send`] is acceptance by the local socket, not
346/// delivery.
347///
348/// [`Drop`] clears the socket before `context` is released, so `zmq_close`
349/// runs while the ZeroMQ context is still alive.
350pub struct ZmqMessagePublisher {
351    socket: Mutex<Option<ExclusiveSocket>>,
352    /// Kept so `zmq_ctx_term` cannot run before [`Drop`] closes `socket`.
353    #[allow(dead_code)]
354    context: zmq::Context,
355}
356
357impl ZmqMessagePublisher {
358    /// Bind a PUB socket. Options from `config` are applied before `bind`.
359    pub fn bind(config: ZmqTransportConfig) -> Result<Self, ZmqMessageError> {
360        let context = shared_context();
361        let socket = open_socket(&context, zmq::PUB, &config, SocketMode::Bind)?;
362        Ok(Self {
363            socket: Mutex::new(Some(socket)),
364            context,
365        })
366    }
367
368    /// Validate `message`, encode one canonical envelope, and send it as one frame.
369    ///
370    /// A closed socket returns [`ZmqMessageError::Closed`] before payload
371    /// validation. An encoded frame larger than [`STRUCTURED_MAX_FRAME_BYTES`]
372    /// returns [`ZmqMessageError::FrameTooLarge`]. The explicit [`Validate`]
373    /// call stays in front of [`encode_canonical_ipc_message`], which validates again.
374    pub fn send(&self, message: &IpcMessage) -> Result<(), ZmqMessageError> {
375        self.require_open()?;
376        message.validate()?;
377        let bytes = encode_canonical_ipc_message(message).map_err(ZmqMessageError::Encode)?;
378        ensure_frame_fits(bytes.len())?;
379        self.send_bytes(&bytes)
380    }
381
382    fn require_open(&self) -> Result<(), ZmqMessageError> {
383        if lock_socket(&self.socket).is_none() {
384            return Err(ZmqMessageError::Closed);
385        }
386        Ok(())
387    }
388
389    fn send_bytes(&self, bytes: &[u8]) -> Result<(), ZmqMessageError> {
390        let mut guard = lock_socket(&self.socket);
391        let socket = guard.as_mut().ok_or(ZmqMessageError::Closed)?;
392        map_send(socket.send(bytes, zmq::DONTWAIT))
393    }
394
395    /// Drop the socket. Idempotent. Further [`Self::send`] calls return [`ZmqMessageError::Closed`].
396    pub fn close(&self) {
397        *lock_socket(&self.socket) = None;
398    }
399
400    /// Live socket options after bind: sndhwm, rcvhwm, linger, reconnect interval, reconnect max, max frame.
401    #[doc(hidden)]
402    pub fn socket_options_for_tests(
403        &self,
404    ) -> Result<(i32, i32, i32, i32, i32, i64), ZmqMessageError> {
405        let guard = lock_socket(&self.socket);
406        let socket = guard.as_ref().ok_or(ZmqMessageError::Closed)?;
407        socket
408            .option_snapshot()
409            .map_err(|err| ZmqMessageError::Transport(err.to_string()))
410    }
411}
412
413impl Drop for ZmqMessagePublisher {
414    fn drop(&mut self) {
415        // Close the socket (`zmq_close`) before the context field is released.
416        *lock_socket(&self.socket) = None;
417    }
418}
419
420impl std::fmt::Debug for ZmqMessageSubscriber {
421    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
422        formatter.write_str("ZmqMessageSubscriber")
423    }
424}
425
426/// Subscribes to one canonical JSON [`IpcMessage`] per ZeroMQ message.
427///
428/// The socket is a `SUB` socket that connects to [`ZmqTransportConfig::endpoint`]
429/// and subscribes to every frame. [`Self::try_recv`] is non-blocking.
430///
431/// [`Drop`] clears the socket before `context` is released, so `zmq_close`
432/// runs while the ZeroMQ context is still alive.
433pub struct ZmqMessageSubscriber {
434    socket: Mutex<Option<ExclusiveSocket>>,
435    /// Kept so `zmq_ctx_term` cannot run before [`Drop`] closes `socket`.
436    #[allow(dead_code)]
437    context: zmq::Context,
438}
439
440impl ZmqMessageSubscriber {
441    /// Connect a SUB socket and subscribe to the full stream.
442    ///
443    /// Options from `config` are applied before `connect`. The subscription is
444    /// installed before `connect` so libzmq replays it on reconnect.
445    pub fn connect(config: ZmqTransportConfig) -> Result<Self, ZmqMessageError> {
446        let context = shared_context();
447        let socket = open_socket(&context, zmq::SUB, &config, SocketMode::Connect)?;
448        Ok(Self {
449            socket: Mutex::new(Some(socket)),
450            context,
451        })
452    }
453
454    /// Receive one frame and decode it with the canonical validation policy.
455    ///
456    /// [`ZmqMessageError::WouldBlock`] means `EAGAIN` (no message). Malformed
457    /// JSON, an incompatible `wire_version`, an invalid payload, and multipart
458    /// or empty framing return typed errors. A failed frame is consumed.
459    ///
460    /// Multipart messages are drained up to eight extra frames. If frames still
461    /// remain, the socket is closed so a later receive cannot resume
462    /// mid-message; that later call returns [`ZmqMessageError::Closed`].
463    pub fn try_recv(&self) -> Result<IpcMessage, ZmqMessageError> {
464        let bytes = self.recv_frame()?;
465        decode_frame(&bytes)
466    }
467
468    fn recv_frame(&self) -> Result<Vec<u8>, ZmqMessageError> {
469        let mut guard = lock_socket(&self.socket);
470        let read = match guard.as_mut() {
471            Some(socket) => read_socket_frame(socket),
472            None => return Err(ZmqMessageError::Closed),
473        };
474        finish_read(&mut guard, read)
475    }
476
477    /// Drop the socket. Idempotent. Further [`Self::try_recv`] calls return [`ZmqMessageError::Closed`].
478    pub fn close(&self) {
479        *lock_socket(&self.socket) = None;
480    }
481
482    /// Test/fuzz hook: the frame decoder [`Self::try_recv`] runs on one
483    /// received single-part frame, without a socket.
484    #[doc(hidden)]
485    pub fn decode_frame_for_tests(bytes: &[u8]) -> Result<IpcMessage, ZmqMessageError> {
486        decode_frame(bytes)
487    }
488
489    pub(super) fn is_closed(&self) -> bool {
490        lock_socket(&self.socket).is_none()
491    }
492
493    /// Live socket options after connect. See [`ZmqMessagePublisher::socket_options_for_tests`].
494    #[doc(hidden)]
495    pub fn socket_options_for_tests(
496        &self,
497    ) -> Result<(i32, i32, i32, i32, i32, i64), ZmqMessageError> {
498        let guard = lock_socket(&self.socket);
499        let socket = guard.as_ref().ok_or(ZmqMessageError::Closed)?;
500        socket
501            .option_snapshot()
502            .map_err(|err| ZmqMessageError::Transport(err.to_string()))
503    }
504}
505
506fn read_frame(socket: &mut ExclusiveSocket) -> Result<Vec<u8>, ZmqMessageError> {
507    match socket.recv_bytes(zmq::DONTWAIT) {
508        Ok(bytes) => Ok(bytes),
509        Err(zmq::Error::EAGAIN) => Err(ZmqMessageError::WouldBlock),
510        Err(err) => Err(ZmqMessageError::Transport(err.to_string())),
511    }
512}
513
514fn more_frames(socket: &mut ExclusiveSocket) -> Result<bool, ZmqMessageError> {
515    socket
516        .rcvmore()
517        .map_err(|err| ZmqMessageError::Transport(err.to_string()))
518}
519
520impl Drop for ZmqMessageSubscriber {
521    fn drop(&mut self) {
522        *lock_socket(&self.socket) = None;
523    }
524}
525
526enum SocketRead {
527    Frame(Vec<u8>),
528    Failed(ZmqMessageError),
529    Abandoned,
530}
531
532fn read_socket_frame(socket: &mut ExclusiveSocket) -> SocketRead {
533    match read_frame(socket) {
534        Ok(bytes) => assemble_frame(socket, bytes),
535        Err(err) => SocketRead::Failed(err),
536    }
537}
538
539fn assemble_frame(socket: &mut ExclusiveSocket, bytes: Vec<u8>) -> SocketRead {
540    match more_frames(socket) {
541        Ok(true) => after_multipart(socket),
542        Ok(false) => SocketRead::Frame(bytes),
543        Err(err) => SocketRead::Failed(err),
544    }
545}
546
547fn after_multipart(socket: &mut ExclusiveSocket) -> SocketRead {
548    match drain_multipart(socket) {
549        Ok(false) => multipart_fault(),
550        Ok(true) => SocketRead::Abandoned,
551        Err(err) => SocketRead::Failed(err),
552    }
553}
554
555fn multipart_fault() -> SocketRead {
556    SocketRead::Failed(ZmqMessageError::UnexpectedFraming(FramingFault::Multipart))
557}
558
559fn finish_read(
560    slot: &mut Option<ExclusiveSocket>,
561    read: SocketRead,
562) -> Result<Vec<u8>, ZmqMessageError> {
563    match read {
564        SocketRead::Frame(bytes) => Ok(bytes),
565        SocketRead::Failed(err) => Err(err),
566        SocketRead::Abandoned => {
567            *slot = None;
568            Err(ZmqMessageError::UnexpectedFraming(FramingFault::Multipart))
569        }
570    }
571}
572
573/// Extra frames drained from one multipart message before the subscriber closes.
574pub(super) const MULTIPART_DRAIN_LIMIT: usize = 8;
575
576fn drain_multipart(socket: &mut ExclusiveSocket) -> Result<bool, ZmqMessageError> {
577    for _ in 0..MULTIPART_DRAIN_LIMIT {
578        if !next_part(socket)? {
579            return Ok(false);
580        }
581    }
582    Ok(true)
583}
584
585fn next_part(socket: &mut ExclusiveSocket) -> Result<bool, ZmqMessageError> {
586    match socket.recv_bytes(zmq::DONTWAIT) {
587        Ok(_) => more_frames(socket),
588        Err(zmq::Error::EAGAIN) => Ok(false),
589        Err(err) => Err(ZmqMessageError::Transport(err.to_string())),
590    }
591}
592
593/// `EnvelopeError::Payload` is every serde payload failure. Structural JSON
594/// (missing fields, wrong types) stays a communication error. A failure is a
595/// validation error only when its text contains a [`ValidationKind`] display
596/// prefix, which is the stable `"{kind} at `" form from [`crate::ValidationError`].
597fn is_validation_payload(err: &serde_json::Error) -> bool {
598    let text = err.to_string();
599    validation_kinds()
600        .iter()
601        .any(|kind| text.contains(marker_prefix(*kind)))
602}
603
604fn validation_kinds() -> [ValidationKind; 6] {
605    [
606        ValidationKind::NonFinite,
607        ValidationKind::OutOfRange,
608        ValidationKind::LengthMismatch,
609        ValidationKind::LimitExceeded,
610        ValidationKind::NestedMetadata,
611        ValidationKind::DuplicateIdentifier,
612    ]
613}
614
615fn marker_prefix(kind: ValidationKind) -> &'static str {
616    match kind {
617        ValidationKind::NonFinite => "non_finite at `",
618        ValidationKind::OutOfRange => "out_of_range at `",
619        ValidationKind::LengthMismatch => "length_mismatch at `",
620        ValidationKind::LimitExceeded => "limit_exceeded at `",
621        ValidationKind::NestedMetadata => "nested_metadata at `",
622        ValidationKind::DuplicateIdentifier => "duplicate_identifier at `",
623    }
624}
625
626pub(super) fn map_transport_error(err: ZmqMessageError) -> crate::BackendError {
627    match err {
628        ZmqMessageError::Validation(err) => crate::BackendError::InvalidInput(err.to_string()),
629        ZmqMessageError::Encode(CanonicalEncodeError::Validation(err)) => {
630            crate::BackendError::InvalidInput(err.to_string())
631        }
632        ZmqMessageError::Envelope(crate::EnvelopeError::Payload(err))
633            if is_validation_payload(&err) =>
634        {
635            crate::BackendError::InvalidInput(err.to_string())
636        }
637        other => crate::BackendError::CommunicationError(other.to_string()),
638    }
639}
640
641// Compile-time Send/Sync record. The raw socket stays `!Sync`; no `unsafe impl Sync`.
642const _: fn() = assert_socket_send_sync;
643
644fn assert_socket_send_sync() {
645    fn assert_send<T: Send>() {}
646    fn assert_sync<T: Sync>() {}
647
648    assert_send::<zmq::Socket>();
649    assert_send::<ExclusiveSocket>();
650    assert_send::<ZmqMessagePublisher>();
651    assert_sync::<ZmqMessagePublisher>();
652    assert_send::<ZmqMessageSubscriber>();
653    assert_sync::<ZmqMessageSubscriber>();
654
655    <zmq::Socket as not_sync::IfSyncThenAmbiguous<_>>::witness();
656    <ExclusiveSocket as not_sync::IfSyncThenAmbiguous<_>>::witness();
657}
658
659/// Overlapping-impl witness that a concrete type is not `Sync`.
660mod not_sync {
661    pub trait IfSyncThenAmbiguous<Marker> {
662        fn witness() {}
663    }
664
665    pub struct NotSyncToken;
666    pub struct SyncToken;
667
668    impl<T: ?Sized> IfSyncThenAmbiguous<NotSyncToken> for T {}
669    impl<T: ?Sized + Sync> IfSyncThenAmbiguous<SyncToken> for T {}
670}
671
672#[cfg(test)]
673#[path = "transport_tests.rs"]
674mod coverage;
corpus_ipc/zmq_message/transport.rs

corpus_ipc/zmq_message/
transport.rs
