rustzmq2 0.1.0

//! Generic socket backend (DEALER / ROUTER / PUSH / PULL).
//!
//! The recv side is a **shared tagged inbound channel**: every
//! `PeerEngine`'s reader task forwards messages through the engine's
//! per-peer inbound channel, and we spawn a small forwarder task per
//! peer that drains that channel into a single `flume::Sender<(PeerIdentity,
//! ...)>` shared across all peers. Socket-layer `recv` pulls from the
//! shared receiver — which is what libzmq's `fq_t` achieves, except
//! without the O(n) scan across peers or the per-poll lock acquisition.
//!
//! Why per-peer forwarder? `PeerEngine::recv` returns `flume::Receiver`
//! clones per peer; we *could* build an async select across them all,
//! but N-way dynamic select in flume is awkward. A tiny forwarder task
//! per peer keeps the fast path one `recv_async` on a single receiver.
//! The cost is one extra task per connection (~1 KB each) and one extra
//! channel hop. Worth it for the simplicity — and for the fact that
//! ROUTER needs the `PeerIdentity` tag attached at point-of-decode,
//! which is most naturally done where the `peer_id` is already in scope.

use crate::codec::{CodecError, FramedIo, IntoEngineWriter, Message};
use crate::endpoint::Endpoint;
use crate::engine::peer_loop::{ConflateSlot, ConflateSlotInner};
#[cfg(feature = "inproc")]
use crate::engine::registry::AnyEngine;
use crate::engine::registry::{make_framed_engine, PeerKey, PeerRegistry};
use crate::engine::PeerEngine;
#[cfg(feature = "inproc")]
use crate::engine::{connect_inproc_engine, inproc_placeholder_engine, InprocInboundTx};
use crate::error::SendError;
use crate::PeerIdentity;
use crate::{
    MultiPeerBackend, SocketBackend, SocketEvent, SocketOptions, SocketType, ZmqError, ZmqMessage,
    ZmqResult,
};

use crate::async_rt::notify::AsyncNotify;
use flume::{Receiver, Sender};
use futures::channel::mpsc;
use parking_lot::Mutex;

use std::collections::HashMap;
use std::sync::Arc;

/// Notifier used by the reconnect task to learn when a peer disconnects.
pub(crate) type DisconnectNotifier = mpsc::Sender<PeerIdentity>;

#[doc(hidden)]
pub struct GenericSocketBackend {
    registry: PeerRegistry,
    inbound_tx: Sender<(
        crate::engine::registry::PeerKey,
        Result<Message, CodecError>,
    )>,
    inbound_rx: Receiver<(
        crate::engine::registry::PeerKey,
        Result<Message, CodecError>,
    )>,
    #[cfg(feature = "inproc")]
    inproc_inbound_tx: InprocInboundTx,
    #[cfg(feature = "inproc")]
    pub(crate) inproc_inbound_rx: crate::engine::InprocInboundRx,
    #[cfg(feature = "inproc")]
    pub(crate) inproc_notify: Arc<crate::async_rt::notify::RuntimeNotify>,
    socket_type: SocketType,
    socket_options: SocketOptions,
    pub(crate) socket_monitor: Mutex<Option<mpsc::Sender<SocketEvent>>>,
    /// Reconnection notifiers keyed by `peer_id`. When a peer disconnects,
    /// we notify the reconnect task so it can attempt reconnection.
    disconnect_notifiers: Mutex<HashMap<PeerIdentity, DisconnectNotifier>>,
    /// Per-peer conflate slots, populated only when `socket_options.conflate = true`.
    /// Each peer gets one slot; `handle_read` overwrites it on every inbound message.
    conflate_slots: Mutex<HashMap<PeerKey, ConflateSlot>>,
    /// Shared notify woken by any peer's `handle_read` in conflate mode.
    pub(crate) conflate_notify: Arc<crate::async_rt::notify::RuntimeNotify>,
}

impl GenericSocketBackend {
    pub(crate) fn with_options(socket_type: SocketType, options: SocketOptions) -> Self {
        let (inbound_tx, inbound_rx) = flume::bounded(options.receive_hwm);
        #[cfg(feature = "inproc")]
        let (inproc_inbound_tx, inproc_inbound_rx) =
            crossbeam_channel::bounded(options.receive_hwm);
        #[cfg(feature = "inproc")]
        let inproc_notify = Arc::new(crate::async_rt::notify::RuntimeNotify::new());
        Self {
            registry: PeerRegistry::new(),
            inbound_tx,
            inbound_rx,
            #[cfg(feature = "inproc")]
            inproc_inbound_tx,
            #[cfg(feature = "inproc")]
            inproc_inbound_rx,
            #[cfg(feature = "inproc")]
            inproc_notify,
            socket_type,
            socket_options: options,
            socket_monitor: Mutex::new(None),
            disconnect_notifiers: Mutex::new(HashMap::new()),
            conflate_slots: Mutex::new(HashMap::new()),
            conflate_notify: Arc::new(crate::async_rt::notify::RuntimeNotify::new()),
        }
    }

    fn build_heartbeat_cfg(opts: &SocketOptions) -> Option<crate::engine::HeartbeatConfig> {
        Some(crate::engine::HeartbeatConfig {
            interval: opts.heartbeat_interval?,
            timeout: opts.heartbeat_timeout?,
            ttl: opts.heartbeat_ttl?,
        })
    }

    /// Register a disconnect notifier for a peer. When the peer disconnects,
    /// the notifier is triggered to wake the reconnect task.
    pub(crate) fn register_disconnect_notifier(
        &self,
        peer_id: PeerIdentity,
        notifier: DisconnectNotifier,
    ) {
        self.disconnect_notifiers.lock().insert(peer_id, notifier);
    }

    /// Receiver side for the socket layer. Consumers call
    /// `recv_async().await` to get the next `(peer_id, message)`.
    pub(crate) fn inbound(
        &self,
    ) -> Receiver<(
        crate::engine::registry::PeerKey,
        Result<Message, CodecError>,
    )> {
        self.inbound_rx.clone()
    }

    /// Exposed so REQ can `registry().next_round_robin()` to pick a
    /// peer for its request, then stash that `peer_id`.
    pub(crate) fn registry(&self) -> &PeerRegistry {
        &self.registry
    }

    /// Like `recv_next` but respects `socket_options.receive_timeout`. Returns
    /// `ZmqError::NoMessage` on timeout (same as a closed channel).
    ///
    /// Reads the inproc receiver and wake-notify handle from `backend`
    /// via `HasInproc` rather than taking them as parameters, so call
    /// sites don't have to cfg-gate the arguments.
    #[cfg(feature = "inproc")]
    pub(crate) async fn recv_next_timed<B: MultiPeerBackend + HasRegistry + HasInproc + ?Sized>(
        inbound: &Receiver<TaggedMsg>,
        backend: &B,
        receive_timeout: Option<std::time::Duration>,
    ) -> ZmqResult<(crate::engine::registry::PeerKey, crate::message::ZmqMessage)> {
        match receive_timeout {
            None => recv_next(inbound, backend).await,
            Some(d) => crate::async_rt::task::timeout(d, recv_next(inbound, backend))
                .await
                .map_err(|_e| ZmqError::NoMessage)?,
        }
    }

    #[cfg(not(feature = "inproc"))]
    pub(crate) async fn recv_next_timed<B: MultiPeerBackend + HasRegistry + ?Sized>(
        inbound: &Receiver<TaggedMsg>,
        backend: &B,
        receive_timeout: Option<std::time::Duration>,
    ) -> ZmqResult<(crate::engine::registry::PeerKey, crate::message::ZmqMessage)> {
        match receive_timeout {
            None => recv_next(inbound, backend).await,
            Some(d) => crate::async_rt::task::timeout(d, recv_next(inbound, backend))
                .await
                .map_err(|_e| ZmqError::NoMessage)?,
        }
    }

    /// Like `recv_next_timed` but uses the conflate per-peer slot path.
    /// Called when `socket_options.conflate = true`. Polls all peer slots;
    /// parks on `conflate_notify` when all are empty.
    pub(crate) async fn recv_next_conflate_timed(
        &self,
        receive_timeout: Option<std::time::Duration>,
    ) -> ZmqResult<(crate::engine::registry::PeerKey, crate::message::ZmqMessage)> {
        let fut = async {
            loop {
                // Snapshot all current slots (under lock, but drop before await).
                let slots: Vec<ConflateSlot> =
                    self.conflate_slots.lock().values().cloned().collect();
                for slot in &slots {
                    if let Some((key, msg)) = slot.slot.lock().take() {
                        return Ok((key, msg));
                    }
                }
                if slots.is_empty() && self.registry.is_empty() {
                    return Err(ZmqError::NoMessage);
                }
                self.conflate_notify.notified().await;
            }
        };
        match receive_timeout {
            None => fut.await,
            Some(d) => crate::async_rt::task::timeout(d, fut)
                .await
                .map_err(|_e| ZmqError::NoMessage)?,
        }
    }

    /// Unified recv: dispatches to the conflate slot path when `conflate=true`,
    /// otherwise to the normal shared-inbound path. Inproc rx/notify are
    /// read from `&self` via `HasInproc`.
    pub(crate) async fn recv_auto(
        &self,
        inbound: &Receiver<TaggedMsg>,
        receive_timeout: Option<std::time::Duration>,
    ) -> ZmqResult<(crate::engine::registry::PeerKey, crate::message::ZmqMessage)> {
        if self.socket_options.conflate {
            return self.recv_next_conflate_timed(receive_timeout).await;
        }
        GenericSocketBackend::recv_next_timed(inbound, self, receive_timeout).await
    }

    /// Round-robin send with optional `send_timeout`. Returns `ZmqError::NoMessage`
    /// on timeout.
    pub(crate) async fn send_round_robin_timed(
        &self,
        message: crate::message::ZmqMessage,
        send_timeout: Option<std::time::Duration>,
    ) -> ZmqResult<PeerIdentity> {
        match send_timeout {
            None => self.send_round_robin(message).await,
            Some(d) => crate::async_rt::task::timeout(d, self.send_round_robin(message))
                .await
                .map_err(|_e| ZmqError::NoMessage)?,
        }
    }

    /// Direct send to a specific peer with optional `send_timeout`.
    pub(crate) async fn send_to_timed(
        &self,
        peer_id: &PeerIdentity,
        message: crate::message::ZmqMessage,
        send_timeout: Option<std::time::Duration>,
    ) -> ZmqResult<()> {
        match send_timeout {
            None => self.send_to(peer_id, message).await,
            Some(d) => crate::async_rt::task::timeout(d, self.send_to(peer_id, message))
                .await
                .map_err(|_e| ZmqError::NoMessage)?,
        }
    }

    /// Round-robin send (DEALER, PUSH). Blocks on HWM backpressure but
    /// returns as soon as the message is enqueued; not when it hits the
    /// wire. Matches libzmq's `zmq_send` contract — pipelined throughput
    /// needs enqueue semantics, not per-message wire-flush semantics.
    pub(crate) async fn send_round_robin(
        &self,
        message: crate::message::ZmqMessage,
    ) -> ZmqResult<PeerIdentity> {
        if self.socket_options.immediate && self.registry.is_empty() {
            return Err(ZmqError::ReturnToSender {
                reason: "Not connected to peers. Unable to send messages".into(),
                message,
            });
        }
        let (key, engine) = match self.registry.next_round_robin() {
            Some(pair) => pair,
            None => {
                return Err(ZmqError::ReturnToSender {
                    reason: "Not connected to peers. Unable to send messages".into(),
                    message,
                })
            }
        };
        match engine.send_msg(message).await {
            Ok(()) => self
                .registry
                .id_for(key)
                .ok_or(ZmqError::Other("Peer disappeared mid-send".into())),
            Err(SendError::Enqueue(Message::Message(m))) => {
                self.registry.remove_by_key(key);
                Err(ZmqError::ReturnToSender {
                    reason: "Not connected to peers. Unable to send messages".into(),
                    message: m,
                })
            }
            Err(SendError::Enqueue(_) | SendError::Flush) => {
                self.registry.remove_by_key(key);
                Err(ZmqError::Other("Peer disconnected during send".into()))
            }
        }
    }

    /// Direct send to a specific peer (ROUTER, REP). Same enqueue
    /// contract as [`GenericSocketBackend::send_round_robin`].
    pub(crate) async fn send_to(
        &self,
        peer_id: &PeerIdentity,
        message: crate::message::ZmqMessage,
    ) -> ZmqResult<()> {
        if self.socket_options.immediate && self.registry.is_empty() {
            return Err(ZmqError::ReturnToSender {
                reason: "Not connected to peers. Unable to send messages".into(),
                message,
            });
        }
        let (key, engine) = self.registry.get_by_id(peer_id).ok_or(ZmqError::Other(
            "Destination client not found by identity".into(),
        ))?;
        match engine.send_msg(message).await {
            Ok(()) => Ok(()),
            Err(SendError::Enqueue(_) | SendError::Flush) => {
                self.registry.remove_by_key(key);
                Err(ZmqError::Other(
                    "Destination client not found by identity".into(),
                ))
            }
        }
    }
}

impl SocketBackend for GenericSocketBackend {
    fn socket_type(&self) -> SocketType {
        self.socket_type
    }

    fn socket_options(&self) -> &SocketOptions {
        &self.socket_options
    }

    fn shutdown(&self) {
        self.registry.clear();
        self.conflate_slots.lock().clear();
    }

    fn monitor(&self) -> &Mutex<Option<mpsc::Sender<SocketEvent>>> {
        &self.socket_monitor
    }
}

/// Small helper trait so `recv_next` can resolve `PeerKey → PeerIdentity`
/// on the cold disconnect path without making every backend leak its
/// registry via a concrete type. All socket backends implement this.
pub(crate) trait HasRegistry {
    fn registry(&self) -> &PeerRegistry;
}

impl HasRegistry for GenericSocketBackend {
    fn registry(&self) -> &PeerRegistry {
        &self.registry
    }
}

/// Companion trait to `HasRegistry`: exposes the inproc inbound receiver
/// and wake-notify handle that backends with inproc support all hold.
/// Lets `recv_next_timed` / `recv_auto` read those fields from `&self`
/// instead of taking them as cfg-gated parameters at every call site.
#[cfg(feature = "inproc")]
pub(crate) trait HasInproc {
    fn inproc_inbound_rx(&self) -> &crossbeam_channel::Receiver<TaggedMsg>;
    fn inproc_notify(&self) -> &Arc<crate::async_rt::notify::RuntimeNotify>;
}

#[cfg(feature = "inproc")]
impl HasInproc for GenericSocketBackend {
    #[inline]
    fn inproc_inbound_rx(&self) -> &crossbeam_channel::Receiver<TaggedMsg> {
        &self.inproc_inbound_rx
    }
    #[inline]
    fn inproc_notify(&self) -> &Arc<crate::async_rt::notify::RuntimeNotify> {
        &self.inproc_notify
    }
}

impl MultiPeerBackend for GenericSocketBackend {
    async fn peer_connected<R, W>(
        self: Arc<Self>,
        peer_id: &PeerIdentity,
        io: FramedIo<R, W>,
        endpoint: Option<Endpoint>,
    ) where
        R: futures::Stream<Item = Result<Message, CodecError>> + Unpin + Send + 'static,
        W: futures::Sink<Message, Error = CodecError> + Unpin + Send + IntoEngineWriter + 'static,
        W::Writer: Send + 'static,
    {
        #[cfg(feature = "curve")]
        let (read_half, write_half, curve) = io.into_parts();
        #[cfg(not(feature = "curve"))]
        let (read_half, write_half) = io.into_parts();
        let inbound_tx = self.inbound_tx.clone();
        let peer_id_owned = peer_id.clone();
        #[cfg(feature = "curve")]
        let mut curve = curve;
        let writer = write_half.into_engine_writer();
        let send_hwm = self.socket_options.send_hwm;
        let conflate = self.socket_options.conflate;
        // Pre-create the slot so we can both pass it into PeerEngine and store it
        // in conflate_slots using the same Arc. The slot is None until the first
        // message arrives from the peer.
        let conflate_slot: Option<ConflateSlot> = if conflate {
            Some(Arc::new(ConflateSlotInner {
                slot: parking_lot::Mutex::new(None),
                notify: self.conflate_notify.clone(),
            }))
        } else {
            None
        };
        let conflate_slot_for_engine = conflate_slot.clone();
        // Resolve the inline-write knob for this engine. The
        // SocketOptions shape encodes user intent:
        //   * outer None → "use type default" (SocketCore::new fills
        //     this in before we get here; treat stray None as disabled
        //     to be safe).
        //   * outer Some(None) → user explicitly disabled.
        //   * outer Some(Some(0)) → enabled, no payload cap.
        //   * outer Some(Some(n)) → enabled, decline payloads >= n.
        // PeerConfig::inline_write_max is `Option<Option<usize>>` where
        // outer = engine-enabled and inner = per-message cap (None =
        // uncapped, Some(n) = cap).
        let inline_write_max: Option<Option<usize>> = match self.socket_options.inline_write_max {
            Some(Some(0)) => Some(None),
            Some(Some(n)) => Some(Some(n)),
            Some(None) | None => None,
        };
        // CURVE engines must skip the inline path — the cipher state
        // lives in peer_loop and inline writes wouldn't go through the
        // encryption stage.
        #[cfg(feature = "curve")]
        let inline_write_max: Option<Option<usize>> = if curve.is_some() {
            None
        } else {
            inline_write_max
        };
        let config = crate::engine::peer_loop::PeerConfig {
            heartbeat: Self::build_heartbeat_cfg(&self.socket_options),
            max_msg_size: self.socket_options.max_msg_size,
            #[cfg(feature = "curve")]
            curve: curve.take(),
            conflate_slot: conflate_slot_for_engine,
            out_batch_size: self.socket_options.out_batch_size,
            // `out_batch_msgs` is `Option<Option<usize>>` on
            // SocketOptions; outer None should never reach here
            // (SocketCore::new fills the per-type default in), but
            // be defensive.
            out_batch_msgs: self.socket_options.out_batch_msgs.unwrap_or(None),
            in_batch_msgs: self.socket_options.in_batch_msgs,
            inline_write_max,
        };
        let (key, _prev) = self.registry.insert_with(peer_id.clone(), |key| {
            make_framed_engine(Arc::new(PeerEngine::spawn(
                key,
                peer_id_owned,
                read_half,
                writer,
                send_hwm,
                inbound_tx,
                config,
            )))
        });
        if let Some(slot) = conflate_slot {
            self.conflate_slots.lock().insert(key, slot);
        }
        // HELLO_MSG: enqueue onto the new peer's outbound so it's the first
        // frame the peer receives (after READY).
        if let Some(hello) = &self.socket_options.hello_msg {
            if let Some((_, engine)) = self.registry.get_by_id(peer_id) {
                let _ = engine.try_send_oneshot(hello.clone());
            }
        }
        // Emit Accepted event for monitor consumers when we have an endpoint.
        if let Some(ep) = endpoint {
            if let Some(tx) = self.socket_monitor.lock().as_mut() {
                let _ = tx.try_send(SocketEvent::Accepted(ep, peer_id.clone()));
            }
        }
    }

    #[cfg(feature = "inproc")]
    #[allow(private_interfaces)]
    async fn peer_connected_inproc(
        self: Arc<Self>,
        peer_id: &PeerIdentity,
        peer: crate::transport::inproc::InprocPeer,
        endpoint: Option<Endpoint>,
    ) -> crate::ZmqResult<()> {
        let inproc_tx = self.inproc_inbound_tx.clone();
        let inproc_notify = self.inproc_notify.clone();
        let (local_key, _prev) = self.registry.insert_with(peer_id.clone(), |_| {
            AnyEngine::Inproc(Arc::new(inproc_placeholder_engine()))
        });
        let local_socket_type = self.socket_type();
        let local_routing_id = self.socket_options.peer_id.clone();
        let (engine, remote_routing_id) = match connect_inproc_engine(
            local_key,
            local_socket_type,
            local_routing_id,
            inproc_tx,
            inproc_notify,
            peer,
        )
        .await
        {
            Ok(pair) => pair,
            Err(e) => {
                self.peer_disconnected(peer_id);
                return Err(e);
            }
        };
        // ROUTER identifies peers by their advertised routing_id, just like
        // over TCP (libzmq `router.cpp:29` sets `recv_routing_id = true`).
        // Swap the placeholder UUID for the remote's id if it provided one.
        let effective_peer_id = if local_socket_type.wants_remote_routing_id() {
            if let Some(remote_id) = remote_routing_id {
                if self.registry.rename_peer_id(local_key, remote_id.clone()) {
                    remote_id
                } else {
                    peer_id.clone()
                }
            } else {
                peer_id.clone()
            }
        } else {
            peer_id.clone()
        };
        self.registry
            .replace_engine(local_key, AnyEngine::Inproc(Arc::new(engine)));
        // HELLO_MSG: match the TCP path above — enqueue onto the new peer's
        // inbound so it's the first message they see. Inproc has no READY
        // frame, so this lands as the first user-level receive.
        if let Some(hello) = &self.socket_options.hello_msg {
            if let Some((_, e)) = self.registry.get_by_id(&effective_peer_id) {
                let _ = e.try_send_oneshot(hello.clone());
            }
        }
        if let Some(ep) = endpoint {
            if let Some(tx) = self.socket_monitor.lock().as_mut() {
                let _ = tx.try_send(SocketEvent::Accepted(ep, effective_peer_id));
            }
        }
        Ok(())
    }

    fn peer_disconnected(&self, peer_id: &PeerIdentity) {
        // DISCONNECT_MSG: best-effort push onto the peer's outbound *before*
        // we remove it from the registry so the writer task still has a
        // channel to drain.
        if let Some(disc) = &self.socket_options.disconnect_msg {
            if let Some((_, engine)) = self.registry.get_by_id(peer_id) {
                let _ = engine.try_send_oneshot(disc.clone());
            }
        }
        if let Some((key, _)) = self.registry.remove_by_id(peer_id) {
            self.conflate_slots.lock().remove(&key);
        }
        if let Some(tx) = self.socket_monitor.lock().as_mut() {
            let _ = tx.try_send(SocketEvent::Disconnected(peer_id.clone()));
        }
        if let Some(mut notifier) = self.disconnect_notifiers.lock().remove(peer_id) {
            let _ = notifier.try_send(peer_id.clone());
        }
    }
}

pub(crate) type TaggedMsg = (
    crate::engine::registry::PeerKey,
    Result<Message, CodecError>,
);

fn handle_tagged_msg<B: MultiPeerBackend + HasRegistry + ?Sized>(
    item: TaggedMsg,
    backend: &B,
) -> Option<ZmqResult<(crate::engine::registry::PeerKey, ZmqMessage)>> {
    match item {
        (peer_key, Ok(Message::Message(m))) => Some(Ok((peer_key, m))),
        (_, Ok(_)) => None,
        (peer_key, Err(CodecError::PeerDisconnected)) => {
            if let Some(id) = backend.registry().id_for(peer_key) {
                backend.peer_disconnected(&id);
            }
            None
        }
        (peer_key, Err(e)) => {
            if let Some(id) = backend.registry().id_for(peer_key) {
                backend.peer_disconnected(&id);
            }
            Some(Err(e.into()))
        }
    }
}

/// Read the next tagged message from the shared inbound, decoding the
/// socket-level `ZmqMessage` out of the wire `Message`. Handles every
/// case the socket-layer recv loop shared across DEALER/ROUTER/REP/
/// PULL/SUB/XPUB:
/// - `Message::Message(m)`: return `(peer_key, m)` — callers that need
///   `PeerIdentity` (ROUTER envelope) look it up via
///   `registry.id_for(key)` themselves.
/// - Non-payload wire frames (greeting/command/shared): skip and loop.
/// - Synthetic `PeerDisconnected` marker from reader EOF: resolve key to
///   identity via the backend's registry, call `peer_disconnected`, loop.
/// - Hard codec error: same, then return the error.
/// - Shared channel closed (socket shutting down): return `NoMessage`.
#[cfg(feature = "inproc")]
pub(crate) async fn recv_next<B>(
    async_inbound: &Receiver<TaggedMsg>,
    backend: &B,
) -> ZmqResult<(crate::engine::registry::PeerKey, ZmqMessage)>
where
    B: MultiPeerBackend + HasRegistry + HasInproc + ?Sized,
{
    use crate::async_rt::notify::AsyncNotify;
    use futures::FutureExt;
    let inproc_inbound = backend.inproc_inbound_rx();
    let inproc_notify = backend.inproc_notify();
    loop {
        // Drain inproc channel first (separate channel, avoids contention with TCP path)
        while let Ok(item) = inproc_inbound.try_recv() {
            if let Some(result) = handle_tagged_msg(item, backend) {
                crate::wake_counter::bump(&crate::wake_counter::RECV_NEXT_WAKES);
                return result;
            }
        }
        // Drain async channel (TCP/IPC peers)
        loop {
            match async_inbound.try_recv() {
                Ok(item) => {
                    if let Some(result) = handle_tagged_msg(item, backend) {
                        crate::wake_counter::bump(&crate::wake_counter::RECV_NEXT_WAKES);
                        return result;
                    }
                }
                Err(flume::TryRecvError::Empty) => break,
                Err(flume::TryRecvError::Disconnected) => return Err(ZmqError::NoMessage),
            }
        }
        // Both empty: park until either wakes us
        futures::select! {
            item = async_inbound.recv_async().fuse() => {
                match item {
                    Ok(item) => {
                        if let Some(result) = handle_tagged_msg(item, backend) {
                            crate::wake_counter::bump(&crate::wake_counter::RECV_NEXT_WAKES);
                            return result;
                        }
                    }
                    Err(_closed) => return Err(ZmqError::NoMessage),
                }
            }
            _ = inproc_notify.notified().fuse() => {
                // inproc channel has data; loop to drain
            }
        }
    }
}

#[cfg(not(feature = "inproc"))]
pub(crate) async fn recv_next<B: MultiPeerBackend + HasRegistry + ?Sized>(
    async_inbound: &Receiver<TaggedMsg>,
    backend: &B,
) -> ZmqResult<(crate::engine::registry::PeerKey, ZmqMessage)> {
    loop {
        match async_inbound.recv_async().await {
            Ok(item) => {
                if let Some(result) = handle_tagged_msg(item, backend) {
                    crate::wake_counter::bump(&crate::wake_counter::RECV_NEXT_WAKES);
                    return result;
                }
            }
            Err(_closed) => return Err(ZmqError::NoMessage),
        }
    }
}