freenet 0.2.42

use anyhow::anyhow;
use dashmap::DashSet;
use either::{Either, Left, Right};
use futures::FutureExt;
use futures::StreamExt;
use std::convert::Infallible;
use std::future::Future;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::pin::Pin;
use std::time::Duration;
use std::{
    collections::{BTreeMap, HashMap, HashSet},
    sync::Arc,
};
use tokio::net::UdpSocket;
use tokio::sync::mpsc::{self, Receiver, Sender, error::TryRecvError};
use tokio::time::Instant;
use tokio::time::{sleep, timeout};
use tracing::Instrument;

use super::{ConnectionError, EventLoopNotificationsReceiver, NetworkBridge};
use crate::contract::{ContractHandlerEvent, WaitingTransaction};
use crate::message::{NetMessageV1, QueryResult};
use crate::node::network_bridge::handshake::{
    Command as HandshakeCommand, CommandSender as HandshakeCommandSender, Event as HandshakeEvent,
    HandshakeHandler,
};
use crate::node::network_bridge::priority_select;
use crate::operations::connect::ConnectMsg;
use crate::ring::Location;
#[cfg(feature = "simulation_tests")]
use crate::ring::PeerKey;
use crate::transport::{
    CongestionControlConfig, ExpectedInboundTracker, PeerConnectionApi, Socket, TransportError,
    TransportKeypair, TransportPublicKey, create_connection_handler,
    global_bandwidth::GlobalBandwidthManager, peer_connection::StreamId,
};
use crate::{
    client_events::ClientId,
    config::GlobalExecutor,
    contract::{
        ContractHandlerChannel, ExecutorToEventLoopChannel, NetworkEventListenerHalve,
        WaitingResolution,
    },
    message::{MessageStats, NetMessage, NodeEvent, Transaction, TransactionType},
    node::{
        NetEventRegister, NodeConfig, OpManager, PeerId, handle_aborted_op,
        process_message_decoupled,
    },
    ring::{KnownPeerKeyLocation, PeerConnectionBackoff, PeerKeyLocation},
    tracing::NetEventLog,
};
use freenet_stdlib::client_api::{ContractResponse, HostResponse};

/// Returns the process RSS (Resident Set Size) in bytes by reading /proc/self/statm.
/// Returns None on non-Linux platforms or if the read fails.
fn get_rss_bytes() -> Option<u64> {
    #[cfg(target_os = "linux")]
    {
        let statm = std::fs::read_to_string("/proc/self/statm").ok()?;
        let rss_pages: u64 = statm.split_whitespace().nth(1)?.parse().ok()?;
        // SAFETY: `sysconf(_SC_PAGESIZE)` is a POSIX-defined, signal-safe
        // call that reads a system constant and has no preconditions.
        let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
        if page_size > 0 {
            Some(rss_pages * page_size as u64)
        } else {
            None
        }
    }
    #[cfg(not(target_os = "linux"))]
    {
        None
    }
}

/// Represents the different ways the event loop can exit.
///
/// This enum is used instead of string-based error matching to provide
/// type-safe handling of shutdown conditions.
#[derive(Debug)]
pub enum EventLoopExitReason {
    /// Graceful shutdown via NodeEvent::Disconnect or ClosedChannel
    GracefulShutdown,
    /// Unexpected stream termination (priority_select returned None)
    UnexpectedStreamEnd,
}

impl std::fmt::Display for EventLoopExitReason {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EventLoopExitReason::GracefulShutdown => write!(f, "Graceful shutdown"),
            EventLoopExitReason::UnexpectedStreamEnd => {
                write!(f, "Network event stream ended unexpectedly")
            }
        }
    }
}

impl std::error::Error for EventLoopExitReason {}

pub(crate) enum P2pBridgeEvent {
    Message(PeerKeyLocation, Box<NetMessage>),
    NodeAction(NodeEvent),
    StreamSend {
        target_addr: SocketAddr,
        stream_id: StreamId,
        data: bytes::Bytes,
        metadata: Option<bytes::Bytes>,
        /// Optional completion signal for broadcast queue concurrency control.
        /// Signaled when the actual stream transfer finishes (not just enqueue).
        completion_tx: Option<tokio::sync::oneshot::Sender<()>>,
    },
    /// Pipe an inbound stream to a target, forwarding fragments incrementally.
    PipeStream {
        target_addr: SocketAddr,
        outbound_stream_id: StreamId,
        inbound_handle: crate::transport::peer_connection::streaming::StreamHandle,
        metadata: Option<bytes::Bytes>,
    },
}

impl std::fmt::Debug for P2pBridgeEvent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Message(peer, msg) => f.debug_tuple("Message").field(peer).field(msg).finish(),
            Self::NodeAction(action) => f.debug_tuple("NodeAction").field(action).finish(),
            Self::StreamSend {
                target_addr,
                stream_id,
                data,
                metadata,
                ..
            } => f
                .debug_struct("StreamSend")
                .field("target_addr", target_addr)
                .field("stream_id", stream_id)
                .field("data_len", &data.len())
                .field("has_metadata", &metadata.is_some())
                .finish(),
            Self::PipeStream {
                target_addr,
                outbound_stream_id,
                ..
            } => f
                .debug_struct("PipeStream")
                .field("target_addr", target_addr)
                .field("outbound_stream_id", outbound_stream_id)
                .finish(),
        }
    }
}

#[derive(Clone)]
pub(crate) struct P2pBridge {
    #[allow(dead_code)]
    accepted_peers: Arc<DashSet<SocketAddr>>,
    ev_listener_tx: Sender<P2pBridgeEvent>,
    op_manager: Arc<OpManager>,
    log_register: Arc<dyn NetEventRegister>,
    /// Configured gateways with their public keys, used for key lookup during reconnection.
    gateways: Arc<Vec<PeerKeyLocation>>,
}

impl P2pBridge {
    fn new<EL>(
        sender: Sender<P2pBridgeEvent>,
        op_manager: Arc<OpManager>,
        event_register: EL,
        gateways: Arc<Vec<PeerKeyLocation>>,
    ) -> Self
    where
        EL: NetEventRegister,
    {
        Self {
            accepted_peers: Arc::new(DashSet::new()),
            ev_listener_tx: sender,
            op_manager,
            log_register: Arc::new(event_register),
            gateways,
        }
    }

    /// Handle orphaned transactions after a peer connection is pruned.
    ///
    /// This triggers retry logic for operations that were pending on the disconnected peer,
    /// allowing them to be routed through alternate peers instead of waiting for timeout.
    pub(crate) async fn handle_orphaned_transactions(
        &self,
        transactions: Vec<Transaction>,
        gateways: &[PeerKeyLocation],
    ) {
        if transactions.is_empty() {
            return;
        }

        tracing::debug!(
            count = transactions.len(),
            "Handling orphaned transactions from pruned connection"
        );

        // Process all orphaned transactions concurrently for better performance
        let results = futures::future::join_all(transactions.into_iter().map(|tx| {
            let op_manager = &self.op_manager;
            async move {
                tracing::debug!(
                    %tx,
                    tx_type = ?tx.transaction_type(),
                    "Attempting retry for orphaned transaction"
                );
                (tx, handle_aborted_op(tx, op_manager, gateways).await)
            }
        }))
        .await;

        // Log any failures
        for (tx, result) in results {
            if let Err(err) = result {
                tracing::warn!(
                    %tx,
                    error = %err,
                    "Failed to handle orphaned transaction"
                );
            }
        }
    }

    /// Send stream data with an explicit completion signal.
    ///
    /// Like `NetworkBridge::send_stream` but allows passing a `completion_tx`
    /// that will be signaled when the actual stream transfer finishes at the
    /// transport layer. Used by the broadcast queue for concurrency control.
    pub(super) async fn send_stream_with_completion(
        &self,
        target_addr: SocketAddr,
        stream_id: StreamId,
        data: bytes::Bytes,
        metadata: Option<bytes::Bytes>,
        completion_tx: Option<tokio::sync::oneshot::Sender<()>>,
    ) -> super::ConnResult<()> {
        self.ev_listener_tx
            .send(P2pBridgeEvent::StreamSend {
                target_addr,
                stream_id,
                data,
                metadata,
                completion_tx,
            })
            .await
            .map_err(|_| ConnectionError::SendNotCompleted(target_addr))?;
        Ok(())
    }
}

impl NetworkBridge for P2pBridge {
    async fn drop_connection(&mut self, peer_addr: SocketAddr) -> super::ConnResult<()> {
        if self.accepted_peers.remove(&peer_addr).is_some() {
            self.ev_listener_tx
                .send(P2pBridgeEvent::NodeAction(NodeEvent::DropConnection(
                    peer_addr,
                )))
                .await
                .map_err(|_| ConnectionError::SendNotCompleted(peer_addr))?;
            // Log disconnect with PeerKeyLocation if we can construct one
            if let Some(peer_loc) = self
                .op_manager
                .ring
                .connection_manager
                .get_peer_by_addr(peer_addr)
            {
                // Peer from connection manager should have known address
                if let Ok(known_loc) = KnownPeerKeyLocation::try_from(&peer_loc) {
                    use crate::tracing::DisconnectReason;
                    // Capture connection duration before the connection is removed
                    let connection_duration_ms = self
                        .op_manager
                        .ring
                        .connection_manager
                        .get_connection_duration_ms(peer_addr);
                    if let Some(event) = NetEventLog::disconnected_with_context(
                        &self.op_manager.ring,
                        &PeerId::new(peer_loc.pub_key().clone(), known_loc.socket_addr()),
                        DisconnectReason::RemoteDropped,
                        connection_duration_ms,
                        None, // bytes_sent not tracked yet
                        None, // bytes_received not tracked yet
                    ) {
                        self.log_register.register_events(Either::Left(event)).await;
                    }
                }
            }
        }
        Ok(())
    }

    async fn send(&self, target_addr: SocketAddr, msg: NetMessage) -> super::ConnResult<()> {
        // Note: outbound event tracing (UnsubscribeSent, etc.) is handled in the
        // OutboundMessageWithTarget handler in the event loop, not here. All messages
        // from P2pBridge::send() flow through handle_bridge_msg → OutboundMessageWithTarget
        // → the shared handler which calls register_events(from_outbound_msg(...)).
        // Tracing here would cause duplicate events.

        // Look up the full PeerKeyLocation from connection manager for transaction tracking
        let target_loc = self
            .op_manager
            .ring
            .connection_manager
            .get_peer_by_addr(target_addr);
        if let Some(ref target) = target_loc {
            self.op_manager.sending_transaction(target, &msg);
            self.ev_listener_tx
                .send(P2pBridgeEvent::Message(target.clone(), Box::new(msg)))
                .await
                .map_err(|_| ConnectionError::SendNotCompleted(target_addr))?;
        } else {
            // No known peer at this address - try to find the public key from gateways
            let gateway_peer = self
                .gateways
                .iter()
                .find(|gw| gw.socket_addr() == Some(target_addr))
                .cloned();

            if let Some(gw_peer) = gateway_peer {
                tracing::debug!(
                    peer_addr = %target_addr,
                    phase = "send",
                    "Using gateway public key for reconnection"
                );
                self.op_manager.sending_transaction(&gw_peer, &msg);
                self.ev_listener_tx
                    .send(P2pBridgeEvent::Message(gw_peer, Box::new(msg)))
                    .await
                    .map_err(|_| ConnectionError::SendNotCompleted(target_addr))?;
            } else {
                // Truly unknown peer - use own key as last resort.
                // This is a normal transient state during connection handshakes
                // when the peer's public key hasn't been established yet.
                tracing::debug!(
                    peer_addr = %target_addr,
                    phase = "send",
                    "Sending to unknown peer address with no known public key"
                );
                let temp_peer = PeerKeyLocation::new(
                    (*self.op_manager.ring.connection_manager.pub_key).clone(),
                    target_addr,
                );
                self.op_manager.sending_transaction(&temp_peer, &msg);
                self.ev_listener_tx
                    .send(P2pBridgeEvent::Message(temp_peer, Box::new(msg)))
                    .await
                    .map_err(|_| ConnectionError::SendNotCompleted(target_addr))?;
            }
        }
        Ok(())
    }

    async fn send_stream(
        &self,
        target_addr: SocketAddr,
        stream_id: StreamId,
        data: bytes::Bytes,
        metadata: Option<bytes::Bytes>,
    ) -> super::ConnResult<()> {
        self.ev_listener_tx
            .send(P2pBridgeEvent::StreamSend {
                target_addr,
                stream_id,
                data,
                metadata,
                completion_tx: None,
            })
            .await
            .map_err(|_| ConnectionError::SendNotCompleted(target_addr))?;
        Ok(())
    }

    async fn pipe_stream(
        &self,
        target_addr: SocketAddr,
        outbound_stream_id: StreamId,
        inbound_handle: crate::transport::peer_connection::streaming::StreamHandle,
        metadata: Option<bytes::Bytes>,
    ) -> super::ConnResult<()> {
        self.ev_listener_tx
            .send(P2pBridgeEvent::PipeStream {
                target_addr,
                outbound_stream_id,
                inbound_handle,
                metadata,
            })
            .await
            .map_err(|_| ConnectionError::SendNotCompleted(target_addr))?;
        Ok(())
    }
}

type PeerConnChannelSender = Sender<Either<NetMessage, ConnEvent>>;
type PeerConnChannelRecv = Receiver<Either<NetMessage, ConnEvent>>;

/// Entry in the connections HashMap, keyed by SocketAddr.
/// The pub_key is learned from the first message received on this connection.
#[derive(Debug)]
struct ConnectionEntry {
    sender: PeerConnChannelSender,
    /// The peer's public key, learned from the first message.
    /// None for transient connections before identity is established.
    pub_key: Option<TransportPublicKey>,
    /// Unique ID for this connection entry. Used to distinguish stale
    /// TransportClosed events from replaced connections (e.g., after identity change).
    connection_id: u64,
    /// When this transport connection was established.
    /// Used for zombie detection: connections not promoted to the ring
    /// within a timeout are considered zombies and dropped.
    created_at: Instant,
}

/// Check whether a transport connection is a zombie: old enough but not
/// promoted to ring, not pending reservation, and not a gateway.
///
/// Gateway connections are exempt below a 1-hour absolute cap because they
/// are intentionally transient (never promoted to ring) but actively needed
/// for routing (#3595).
///
/// Two thresholds for non-gateway connections:
/// Zombie thresholds are derived from `transient_ttl` (configurable, default 30s):
///
/// - `zombie_threshold` = `transient_ttl * 3`: catches connections with no pending
///   reservation. Must be greater than `PENDING_RESERVATION_TTL` (60s) so that a
///   connection isn't immediately killed after its reservation expires. Previous
///   hardcoded value of 300s caused gateways to accumulate ~250 zombie transports,
///   overwhelming the packet processing channel and dropping keepalive packets.
/// - `absolute_zombie_threshold` = `transient_ttl * 6`: overrides `has_pending` to
///   break the refresh cycle where `connection_maintenance()` perpetually renews
///   pending reservations on gateway transports. Previous hardcoded value of 600s
///   allowed zombie transports to linger far too long.
fn is_zombie(
    created_at_elapsed: Duration,
    in_ring: bool,
    has_pending: bool,
    is_gateway: bool,
    transient_ttl: Duration,
) -> bool {
    let zombie_threshold = transient_ttl * 3;
    let absolute_zombie_threshold = transient_ttl * 6;

    if in_ring {
        return false;
    }
    // Gateway transient connections are intentionally not promoted to ring,
    // but the node needs them for routing. Without this exemption, gateways
    // enter a zombie→prune→reconnect→zombie death spiral that breaks all
    // streaming transfers (#3595).
    //
    // The exemption is time-bounded: truly dead gateway connections (no
    // traffic for 1 hour) are still cleaned up. The transport-level idle
    // timeout is the primary backstop, but this ensures no permanent leaks.
    /// Gateway connections get a generous exemption window (1 hour) because they
    /// are intentionally transient but needed for routing. The transport-level
    /// idle timeout (120s keepalive) is the primary cleanup mechanism for dead
    /// gateways; this threshold is the safety net.
    const GATEWAY_ZOMBIE_EXEMPTION: Duration = Duration::from_secs(3600);
    if is_gateway && created_at_elapsed < GATEWAY_ZOMBIE_EXEMPTION {
        return false;
    }
    if created_at_elapsed > zombie_threshold && !has_pending {
        return true;
    }
    if created_at_elapsed > absolute_zombie_threshold {
        return true;
    }
    false
}

/// Monotonically increasing counter for generating unique connection IDs.
static NEXT_CONNECTION_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

fn next_connection_id() -> u64 {
    NEXT_CONNECTION_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
}

pub(in crate::node) struct P2pConnManager {
    pub(in crate::node) gateways: Vec<PeerKeyLocation>,
    pub(in crate::node) bridge: P2pBridge,
    conn_bridge_rx: Receiver<P2pBridgeEvent>,
    event_listener: Box<dyn NetEventRegister>,
    /// Connections indexed by socket address (the transport-level identifier).
    /// This is the source of truth for active connections.
    /// Uses BTreeMap for deterministic iteration order in simulation tests.
    connections: BTreeMap<SocketAddr, ConnectionEntry>,
    /// Reverse lookup: public key -> socket address.
    /// Used to find connections when we only know the peer's identity.
    /// Must be kept in sync with `connections`.
    /// Uses BTreeMap for deterministic iteration order in simulation tests.
    addr_by_pub_key: BTreeMap<TransportPublicKey, SocketAddr>,
    conn_event_tx: Option<Sender<ConnEvent>>,
    key_pair: TransportKeypair,
    listening_ip: IpAddr,
    listening_port: u16,
    is_gateway: bool,
    /// If set, will sent the location over network messages.
    ///
    /// It will also determine whether to trust the location of peers sent in network messages or derive them from IP.
    ///
    /// This is used for testing deterministically with given location. In production this should always be none
    /// and locations should be derived from IP addresses.
    this_location: Option<Location>,
    check_version: bool,
    bandwidth_limit: Option<usize>,
    /// Global bandwidth manager for fair sharing across all connections.
    global_bandwidth: Option<Arc<GlobalBandwidthManager>>,
    /// Minimum ssthresh floor for LEDBAT timeout recovery.
    ledbat_min_ssthresh: Option<usize>,
    /// Congestion control configuration.
    congestion_config: CongestionControlConfig,
    blocked_addresses: Option<HashSet<SocketAddr>>,
    /// Per-contract retry count for broadcasts that found no targets yet.
    broadcast_retries: HashMap<freenet_stdlib::prelude::ContractKey, u8>,
    /// Tracks how many consecutive broadcast cycles found zero targets per contract.
    /// Used to suppress repetitive WARN logs after the first few failures.
    /// Bounded to MAX_BROADCAST_STREAK_ENTRIES to prevent unbounded growth from
    /// network-influenced contract keys.
    broadcast_no_target_streak: HashMap<freenet_stdlib::prelude::ContractKey, u32>,
    /// Global broadcast queue that serializes outbound broadcast streams
    /// with bounded concurrency to prevent uplink saturation (issue #3337).
    #[cfg(not(feature = "simulation_tests"))]
    broadcast_queue: super::broadcast_queue::BroadcastQueue,
}

impl P2pConnManager {
    pub(in crate::node) fn listening_port(&self) -> u16 {
        self.listening_port
    }

    pub async fn build(
        config: &NodeConfig,
        op_manager: Arc<OpManager>,
        event_listener: impl NetEventRegister + Clone,
    ) -> anyhow::Result<Self> {
        let listen_port = config.network_listener_port;
        let listener_ip = config.network_listener_ip;

        let gateways = config.get_gateways()?;
        let gateways_arc = Arc::new(gateways.clone());
        let key_pair = config.key_pair.clone();

        // Bridge channel capacity: must exceed the max number of connected peers
        // to avoid self-deadlock when the event loop broadcasts inline. The primary
        // mitigation is spawning fan-out via broadcast_to_peers(), but this provides
        // defense-in-depth. Do NOT fan-out more than this many sends inline on the
        // event loop — use broadcast_to_peers() instead.
        let (tx_bridge_cmd, rx_bridge_cmd) = mpsc::channel(512);
        let bridge = P2pBridge::new(
            tx_bridge_cmd,
            op_manager,
            event_listener.clone(),
            gateways_arc,
        );

        // Initialize our peer identity before any connection attempts so join requests can
        // reference the correct address.
        // Note: For peers behind NAT, this initial address is a fallback - the correct
        // external address is learned via ObservedAddress and updated via set_own_addr().
        // The ring location should be computed from the observed external address,
        // not from this initial fallback.
        let advertised_addr = {
            let advertised_ip = config
                .own_addr
                .map(|addr| addr.ip())
                .or(config.config.network_api.public_address)
                .unwrap_or_else(|| {
                    if listener_ip.is_unspecified() {
                        IpAddr::V4(Ipv4Addr::LOCALHOST)
                    } else {
                        listener_ip
                    }
                });
            let advertised_port = config
                .own_addr
                .map(|addr| addr.port())
                .or(config.config.network_api.public_port)
                .unwrap_or(listen_port);
            SocketAddr::new(advertised_ip, advertised_port)
        };
        bridge
            .op_manager
            .ring
            .connection_manager
            .try_set_own_addr(advertised_addr);

        Ok(P2pConnManager {
            gateways,
            bridge,
            conn_bridge_rx: rx_bridge_cmd,
            event_listener: Box::new(event_listener),
            connections: BTreeMap::new(),
            addr_by_pub_key: BTreeMap::new(),
            conn_event_tx: None,
            key_pair,
            listening_ip: listener_ip,
            listening_port: listen_port,
            is_gateway: config.is_gateway,
            this_location: config.location,
            check_version: !config.config.network_api.ignore_protocol_version,
            bandwidth_limit: config.config.network_api.bandwidth_limit,
            global_bandwidth: config
                .config
                .network_api
                .total_bandwidth_limit
                .map(|total| {
                    Arc::new(GlobalBandwidthManager::new(
                        total,
                        config.config.network_api.min_bandwidth_per_connection,
                    ))
                }),
            ledbat_min_ssthresh: config.config.network_api.ledbat_min_ssthresh,
            congestion_config: config.config.network_api.build_congestion_config(),
            blocked_addresses: config.blocked_addresses.clone(),
            broadcast_retries: HashMap::new(),
            broadcast_no_target_streak: HashMap::new(),
            #[cfg(not(feature = "simulation_tests"))]
            broadcast_queue: super::broadcast_queue::BroadcastQueue::new(),
        })
    }

    /// Runs the event listener with `UdpSocket` (production mode).
    ///
    /// This is a convenience wrapper around `run_event_listener_with_socket<UdpSocket>`.
    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(name = "network_event_listener", fields(peer = %self.bridge.op_manager.ring.connection_manager.pub_key), skip_all)]
    pub async fn run_event_listener(
        self,
        op_manager: Arc<OpManager>,
        client_wait_for_transaction: ContractHandlerChannel<WaitingResolution>,
        notification_channel: EventLoopNotificationsReceiver,
        executor_listener: ExecutorToEventLoopChannel<NetworkEventListenerHalve>,
        node_controller: Receiver<NodeEvent>,
    ) -> anyhow::Result<Infallible> {
        self.run_event_listener_with_socket::<UdpSocket>(
            op_manager,
            client_wait_for_transaction,
            notification_channel,
            executor_listener,
            node_controller,
        )
        .await
    }

    /// Runs the event listener with a configurable socket type.
    ///
    /// This is the generic version that allows using different socket implementations:
    /// - `UdpSocket` for production
    /// - `InMemorySocket` for testing
    ///
    /// The socket type only affects connection establishment. Once connections are
    /// established, they are type-erased via `PeerConnectionApi` trait objects.
    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(name = "network_event_listener", fields(peer = %self.bridge.op_manager.ring.connection_manager.pub_key), skip_all)]
    pub async fn run_event_listener_with_socket<S: Socket>(
        self,
        op_manager: Arc<OpManager>,
        client_wait_for_transaction: ContractHandlerChannel<WaitingResolution>,
        notification_channel: EventLoopNotificationsReceiver,
        executor_listener: ExecutorToEventLoopChannel<NetworkEventListenerHalve>,
        node_controller: Receiver<NodeEvent>,
    ) -> anyhow::Result<Infallible> {
        // Destructure self to avoid partial move issues
        let P2pConnManager {
            gateways,
            bridge,
            conn_bridge_rx,
            event_listener,
            connections,
            addr_by_pub_key,
            conn_event_tx: _,
            key_pair,
            listening_ip,
            listening_port,
            is_gateway,
            this_location,
            check_version,
            bandwidth_limit,
            global_bandwidth,
            ledbat_min_ssthresh,
            congestion_config,
            blocked_addresses,
            broadcast_retries,
            broadcast_no_target_streak,
            #[cfg(not(feature = "simulation_tests"))]
            broadcast_queue,
        } = self;

        let (outbound_conn_handler, inbound_conn_handler, mut listen_task_handle) =
            create_connection_handler::<S>(
                key_pair.clone(),
                listening_ip,
                listening_port,
                is_gateway,
                bandwidth_limit,
                global_bandwidth,
                ledbat_min_ssthresh,
                Some(congestion_config.clone()),
            )
            .await?;

        tracing::info!(
            listening_port,
            listening_ip = %listening_ip,
            is_gateway,
            peer_key = %key_pair.public(),
            phase = "startup",
            "Network listener opened - ready to receive connections"
        );

        let expected_inbound = outbound_conn_handler.expected_inbound_tracker();
        let mut state = EventListenerState::new(expected_inbound);

        let (conn_event_tx, conn_event_rx) = mpsc::channel(1024);

        // For non-gateway peers, pass the peer_ready flag so it can be set after first handshake
        // For gateways, pass None (they're always ready)
        let peer_ready = if !is_gateway {
            Some(bridge.op_manager.peer_ready.clone())
        } else {
            None
        };

        let (handshake_handler, handshake_cmd_sender) = HandshakeHandler::new(
            inbound_conn_handler,
            outbound_conn_handler.clone(),
            bridge.op_manager.ring.connection_manager.clone(),
            bridge.op_manager.ring.router.clone(),
            this_location,
            is_gateway,
            peer_ready,
        );

        let select_stream = priority_select::ProductionPrioritySelectStream::new(
            notification_channel.notifications_receiver,
            notification_channel.op_execution_receiver,
            conn_bridge_rx,
            handshake_handler,
            node_controller,
            client_wait_for_transaction,
            executor_listener,
            conn_event_rx,
        );

        // Pin the stream on the stack
        tokio::pin!(select_stream);

        // Reconstruct a P2pConnManager-like structure for use in the loop
        // We can't use the original self because we moved conn_bridge_rx
        let mut ctx = P2pConnManager {
            gateways,
            bridge,
            conn_bridge_rx: tokio::sync::mpsc::channel(1).1, // Dummy, won't be used
            event_listener,
            connections,
            addr_by_pub_key,
            conn_event_tx: Some(conn_event_tx.clone()),
            key_pair,
            listening_ip,
            listening_port,
            is_gateway,
            this_location,
            check_version,
            bandwidth_limit,
            global_bandwidth: None, // Already used for connection handler, not needed in ctx
            ledbat_min_ssthresh,
            congestion_config, // Already used for connection handler, kept for struct completeness
            blocked_addresses,
            broadcast_retries,
            broadcast_no_target_streak,
            #[cfg(not(feature = "simulation_tests"))]
            broadcast_queue: broadcast_queue.clone(),
        };

        // Start the broadcast queue worker (production only).
        // In simulation tests, broadcasts are sent inline for deterministic ordering.
        #[cfg(not(feature = "simulation_tests"))]
        let _broadcast_worker_handle =
            broadcast_queue.start_worker(ctx.bridge.clone(), op_manager.clone());

        // Track whether we exit via graceful shutdown (Disconnect or ClosedChannel)
        // vs unexpected stream end
        let mut graceful_shutdown = false;

        // Event loop instrumentation
        let mut loop_iteration_count = 0u64;
        let mut slow_event_count = 0u64;
        let mut last_stats_log = Instant::now();
        const STATS_LOG_INTERVAL: Duration = Duration::from_secs(30);
        const SLOW_EVENT_THRESHOLD: Duration = Duration::from_millis(100);

        // Monitor both the event stream AND the UDP listen task.
        // If the listen task exits for any reason, the transport layer is dead
        // and we must propagate the error to trigger node shutdown.
        loop {
            let result = tokio::select! {
                biased;

                listen_result = &mut listen_task_handle => {
                    // The UDP listener task has exited — this is fatal.
                    // Without the listener, no UDP packets are read from the socket,
                    // connections will time out, and the node becomes unresponsive.
                    match listen_result {
                        Ok(e) => {
                            tracing::error!(
                                error = %e,
                                "CRITICAL: UDP listen task exited with transport error"
                            );
                        }
                        Err(join_err) => {
                            let cause = if join_err.is_panic() { "panicked" } else { "cancelled" };
                            tracing::error!(
                                cause,
                                "CRITICAL: UDP listen task failed: {join_err}"
                            );
                        }
                    }
                    return Err(anyhow!(
                        "UDP listen task exited unexpectedly — \
                         transport layer is dead, node must restart"
                    ));
                },

                maybe_result = StreamExt::next(&mut select_stream) => {
                    let Some(result) = maybe_result else {
                        break;
                    };
                    result
                },
            };

            loop_iteration_count += 1;

            let event_type = match &result {
                priority_select::SelectResult::Notification(_) => "notification",
                priority_select::SelectResult::OpExecution(_) => "op_execution",
                priority_select::SelectResult::PeerConnection(_) => "peer_connection",
                priority_select::SelectResult::ConnBridge(_) => "conn_bridge",
                priority_select::SelectResult::Handshake(_) => "handshake",
                priority_select::SelectResult::NodeController(_) => "node_controller",
                priority_select::SelectResult::ClientTransaction(_) => "client_transaction",
                priority_select::SelectResult::ExecutorTransaction(_) => "executor_transaction",
            };

            let process_start = Instant::now();

            // Process the result using the existing handler
            let event = ctx
                .process_select_result(result, &mut state, &handshake_cmd_sender)
                .await?;

            let elapsed = process_start.elapsed();
            if elapsed > SLOW_EVENT_THRESHOLD {
                slow_event_count += 1;
                tracing::warn!(
                    event_type,
                    elapsed_ms = elapsed.as_millis(),
                    "Slow event loop iteration"
                );
            }

            // Periodic stats logging
            if last_stats_log.elapsed() > STATS_LOG_INTERVAL {
                let notifier = &op_manager.to_event_listener;
                let transport_connections = ctx.connections.len();
                let ring_connections = op_manager.ring.connection_manager.connection_count();
                tracing::info!(
                    iterations = loop_iteration_count,
                    slow_events = slow_event_count,
                    notification_channel_pending = notifier.notification_channel_pending(),
                    notification_channel_capacity = notifier.notifications_sender.capacity(),
                    active_connections = transport_connections,
                    ring_connections,
                    "Event loop stats"
                );
                // Detect transport/ring divergence: transport has connections but ring doesn't
                if transport_connections > 0 && ring_connections == 0 {
                    tracing::error!(
                        transport_connections,
                        ring_connections,
                        "RING_TRANSPORT_DESYNC: transport has connections but ring topology is empty - \
                         connections are not being promoted or are being immediately pruned"
                    );
                }

                // Periodic ReadyState re-broadcast: if we are ready and have readiness
                // gating enabled, re-broadcast our ReadyState to all peers every 30s.
                // This ensures lost ReadyState messages are recovered within one tick.
                if op_manager.ring.connection_manager.min_ready_connections > 0
                    && op_manager.ring.connection_manager.is_self_ready()
                {
                    ctx.handle_broadcast_ready_state(true).await;
                }

                #[cfg(all(unix, feature = "jemalloc-prof"))]
                {
                    use tikv_jemalloc_ctl::{epoch, stats};
                    epoch::advance().ok();
                    let allocated = stats::allocated::read().unwrap_or(0);
                    let resident = stats::resident::read().unwrap_or(0);
                    tracing::info!(
                        allocated_mb = allocated / (1024 * 1024),
                        resident_mb = resident / (1024 * 1024),
                        "jemalloc memory stats"
                    );
                }

                loop_iteration_count = 0;
                slow_event_count = 0;
                last_stats_log = Instant::now();

                // Zombie transport cleanup: remove connections older than 3× transient_ttl
                // that haven't been promoted to ring and have no pending reservation.
                // An absolute threshold of 6× transient_ttl overrides pending reservations
                // to catch gateway transports stuck in a pending-refresh cycle.
                //
                // IMPORTANT: We use drop_zombie_connection (non-blocking try_send)
                // instead of drop_connection_by_addr to avoid a circular deadlock
                // with the handshake driver (#3519). We also cap the batch size to
                // limit event loop latency — each zombie cleanup involves topology
                // pruning and orphaned transaction handling. With a 100ms timeout
                // per zombie, 64 zombies = ~6.4s worst case per cycle.
                // Remaining zombies will be cleaned up in the next 30s cycle.
                const MAX_ZOMBIE_CLEANUP_PER_CYCLE: usize = 64;
                let transient_ttl = op_manager.ring.connection_manager.transient_ttl();
                let zombie_addrs: Vec<SocketAddr> = ctx
                    .connections
                    .iter()
                    .filter(|(addr, entry)| {
                        let is_gateway = ctx
                            .gateways
                            .iter()
                            .any(|gw| gw.socket_addr() == Some(**addr));
                        is_zombie(
                            entry.created_at.elapsed(),
                            op_manager.ring.connection_manager.is_in_ring(**addr),
                            op_manager
                                .ring
                                .connection_manager
                                .has_connection_or_pending(**addr),
                            is_gateway,
                            transient_ttl,
                        )
                    })
                    .map(|(addr, _)| *addr)
                    .take(MAX_ZOMBIE_CLEANUP_PER_CYCLE)
                    .collect();
                if !zombie_addrs.is_empty() {
                    tracing::info!(
                        zombie_count = zombie_addrs.len(),
                        "Cleaning up zombie transports (not promoted to ring)"
                    );
                }
                for addr in &zombie_addrs {
                    ctx.drop_zombie_connection(*addr, &handshake_cmd_sender)
                        .await;
                }

                // Periodic cleanup of pending_op_results: remove entries where the
                // receiver has been dropped (closed sender). This is a safety net for
                // transactions that complete without emitting TransactionCompleted/TimedOut
                // events, preventing unbounded HashMap growth.
                const PENDING_OP_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
                if state.last_pending_op_cleanup.elapsed() > PENDING_OP_CLEANUP_INTERVAL {
                    let before = state.pending_op_results.len();
                    state
                        .pending_op_results
                        .retain(|_tx, sender| !sender.is_closed());
                    let removed = before - state.pending_op_results.len();
                    if removed > 0 {
                        for _ in 0..removed {
                            crate::config::GlobalTestMetrics::record_pending_op_remove();
                        }
                        crate::config::GlobalTestMetrics::record_pending_op_size(
                            state.pending_op_results.len() as u64,
                        );
                        tracing::info!(
                            removed,
                            remaining = state.pending_op_results.len(),
                            "Cleaned up closed pending_op_results senders"
                        );
                    }
                    state.last_pending_op_cleanup = Instant::now();
                }
            }

            match event {
                EventResult::Continue => continue,
                EventResult::Event(event) => {
                    match *event {
                        ConnEvent::InboundMessage(inbound) => {
                            let remote = inbound.remote_addr;
                            let msg = inbound.msg;
                            tracing::debug!(
                                tx = %msg.id(),
                                msg_type = %msg,
                                peer_addr = ?remote,
                                direction = "inbound",
                                phase = "receive",
                                "Received inbound message from peer"
                            );
                            // NOTE: Do NOT fill in the joiner's address here!
                            // The connect::handle_request() function handles address discovery
                            // and emits ObservedAddress to inform the joiner of their external
                            // address. If we fill it in here, handle_request() won't know that
                            // WE discovered the address and won't send ObservedAddress.
                            // The source_addr is passed to handle_inbound_message and propagates
                            // to the connect operation via source_addr parameter.
                            // Pass the source address through to operations for routing.
                            // This replaces the old rewrite_sender_addr hack - instead of mutating
                            // message contents, we pass the observed transport address separately.
                            ctx.handle_inbound_message(msg, remote, &op_manager, &mut state)
                                .await?;
                        }
                        ConnEvent::OutboundMessage(NetMessage::V1(NetMessageV1::Aborted(tx))) => {
                            // TODO: handle aborted transaction as internal message
                            tracing::warn!(tx = %tx, phase = "abort", "Transaction aborted");
                        }
                        ConnEvent::OutboundMessage(msg) => {
                            // With hop-by-hop routing, messages should go through OutboundMessageWithTarget.
                            // This path should not be reached - if we get here, it means a message was sent
                            // without proper target extraction from operation state.
                            tracing::error!(
                                tx = %msg.id(),
                                msg_type = %msg,
                                phase = "send",
                                "Outbound message missing target - bug in routing logic, processing locally as fallback"
                            );
                            // Process locally as a fallback
                            ctx.handle_inbound_message(msg, None, &op_manager, &mut state)
                                .await?;
                        }
                        ConnEvent::OutboundMessageWithTarget { target_addr, msg } => {
                            // This variant uses an explicit target address from P2pBridge::send(),
                            // which is critical for NAT scenarios where the address in the message
                            // differs from the actual transport address we should send to.
                            tracing::debug!(
                                tx = %msg.id(),
                                msg_type = %msg,
                                peer_addr = %target_addr,
                                direction = "outbound",
                                phase = "send",
                                "Sending outbound message to peer"
                            );

                            // Trace outbound events (UnsubscribeSent, etc.) for telemetry.
                            // Without this, messages routed via handle_notification_msg are
                            // invisible to the event aggregator.
                            ctx.bridge
                                .log_register
                                .register_events(NetEventLog::from_outbound_msg(
                                    &msg,
                                    &ctx.bridge.op_manager.ring,
                                    Some(target_addr),
                                ))
                                .await;

                            // Look up the connection using the explicit target address
                            let peer_connection = ctx.connections.get(&target_addr);

                            match peer_connection {
                                Some(peer_connection) => {
                                    // Short timeout avoids head-of-line blocking: if
                                    // the channel doesn't drain within 500ms the peer
                                    // is congested/dead, so we drop the message and
                                    // let the op timeout + retry. 500ms gives transient
                                    // bursts and slow CI runners time to clear while
                                    // preventing the event loop from stalling on dead
                                    // peers (which would block indefinitely).
                                    const SEND_TIMEOUT: std::time::Duration =
                                        std::time::Duration::from_millis(500);
                                    let msg_type = match &msg {
                                        NetMessage::V1(v1) => match v1 {
                                            NetMessageV1::Connect(_) => "Connect",
                                            NetMessageV1::Put(_) => "Put",
                                            NetMessageV1::Get(_) => "Get",
                                            NetMessageV1::Subscribe(_) => "Subscribe",
                                            NetMessageV1::Update(_) => "Update",
                                            NetMessageV1::Aborted(_) => "Aborted",
                                            NetMessageV1::NeighborHosting { .. } => {
                                                "NeighborHosting"
                                            }
                                            NetMessageV1::InterestSync { .. } => "InterestSync",
                                            NetMessageV1::ReadyState { .. } => "ReadyState",
                                        },
                                    };
                                    match tokio::time::timeout(
                                        SEND_TIMEOUT,
                                        peer_connection.sender.send(Left(msg.clone())),
                                    )
                                    .await
                                    {
                                        Ok(Ok(())) => {
                                            tracing::trace!(
                                                tx = %msg.id(),
                                                peer_addr = %target_addr,
                                                phase = "send",
                                                "Message sent to peer connection"
                                            );
                                        }
                                        Ok(Err(_closed)) => {
                                            tracing::error!(
                                                tx = %msg.id(),
                                                peer_addr = %target_addr,
                                                msg_type,
                                                phase = "error",
                                                "Peer connection channel closed"
                                            );
                                        }
                                        Err(_timeout) => {
                                            tracing::warn!(
                                                tx = %msg.id(),
                                                peer_addr = %target_addr,
                                                msg_type,
                                                phase = "backpressure",
                                                "Outbound channel full after {SEND_TIMEOUT:?}, \
                                                 dropping message to avoid event loop stall"
                                            );
                                        }
                                    }
                                }
                                None => {
                                    // No existing connection - need to establish one first.
                                    // This happens for initial Connect requests to a gateway.
                                    let tx = *msg.id();

                                    // Try to get full peer info from operation state (needed for handshake)
                                    // Uses peek method to avoid pop/push overhead
                                    let target_peer = ctx.bridge.op_manager.peek_target_peer(&tx);

                                    // If peek_target_peer returns None (e.g., for Put/Get operations which
                                    // only store addresses, not full peer info), try looking up the peer
                                    // in the connection_manager by address. This enables response routing
                                    // when the connection still exists but operation state lacks full peer info.
                                    let target_peer = target_peer.or_else(|| {
                                        ctx.bridge
                                            .op_manager
                                            .ring
                                            .connection_manager
                                            .get_peer_location_by_addr(target_addr)
                                    });

                                    // If still not found, check if this is a known gateway.
                                    // Gateways have their public keys configured at startup.
                                    let target_peer = target_peer.or_else(|| {
                                        ctx.gateways
                                            .iter()
                                            .find(|gw| gw.socket_addr() == Some(target_addr))
                                            .cloned()
                                    });

                                    let Some(target_peer) = target_peer else {
                                        // Can't establish connection without peer info.
                                        // This happens when the connection was dropped before we could
                                        // route the response. Since the peer is no longer in
                                        // connection_manager, we can't re-establish the connection.
                                        // Log at warn level since this can legitimately happen with
                                        // transient connections that time out before response routing.
                                        tracing::warn!(
                                            tx = %tx,
                                            peer_addr = %target_addr,
                                            active_connections = ctx.connections.len(),
                                            phase = "connect",
                                            "Cannot establish connection - peer not found in connection_manager. \
                                             Connection likely dropped before response could be routed."
                                        );
                                        ctx.bridge.op_manager.completed(tx);
                                        continue;
                                    };

                                    tracing::debug!(
                                        tx = %tx,
                                        peer_addr = %target_addr,
                                        peer = %target_peer,
                                        active_connections = ctx.connections.len(),
                                        phase = "connect",
                                        "No existing connection - establishing connection before sending message"
                                    );

                                    // Queue the message for sending after connection is established
                                    let (callback, mut result) = mpsc::channel(10);
                                    let msg_clone = msg.clone();
                                    let bridge_sender = ctx.bridge.ev_listener_tx.clone();
                                    let op_manager = ctx.bridge.op_manager.clone();
                                    let gateways = ctx.gateways.clone();

                                    // Mark gateway connections as transient to prevent
                                    // premature ring promotion (should only happen via
                                    // ConnectResponse).
                                    let is_gw = gateways
                                        .iter()
                                        .any(|gw| gw.socket_addr() == Some(target_addr));
                                    ctx.bridge
                                        .ev_listener_tx
                                        .send(P2pBridgeEvent::NodeAction(NodeEvent::ConnectPeer {
                                            peer: target_peer.clone(),
                                            tx,
                                            callback,
                                            is_gw,
                                        }))
                                        .await?;

                                    tracing::debug!(
                                        tx = %tx,
                                        peer_addr = %target_addr,
                                        phase = "connect",
                                        "Dispatched ConnectPeer event - waiting for connection establishment"
                                    );

                                    // Spawn a task to wait for connection and then send the message
                                    let target_peer_for_resend = target_peer.clone();
                                    GlobalExecutor::spawn(async move {
                                        match timeout(Duration::from_secs(20), result.recv()).await
                                        {
                                            Ok(Some(Ok((connected_addr, _)))) => {
                                                tracing::debug!(
                                                    tx = %tx,
                                                    peer_addr = %connected_addr,
                                                    phase = "connect",
                                                    "Connection established - rescheduling message send"
                                                );
                                                // Construct PeerKeyLocation with the connected address
                                                // and the original public key
                                                let connected_peer = PeerKeyLocation::new(
                                                    target_peer_for_resend.pub_key().clone(),
                                                    connected_addr,
                                                );
                                                // Re-send via P2pBridge which will route correctly
                                                if let Err(e) = bridge_sender
                                                    .send(P2pBridgeEvent::Message(
                                                        connected_peer,
                                                        Box::new(msg_clone),
                                                    ))
                                                    .await
                                                {
                                                    tracing::error!(
                                                        tx = %tx,
                                                        peer_addr = %connected_addr,
                                                        error = ?e,
                                                        phase = "error",
                                                        "Failed to reschedule message after connection established"
                                                    );
                                                }
                                            }
                                            Ok(Some(Err(e))) => {
                                                tracing::debug!(
                                                    tx = %tx,
                                                    peer_addr = %target_addr,
                                                    error = ?e,
                                                    phase = "error",
                                                    "Connection attempt failed"
                                                );
                                                if let Err(err) =
                                                    handle_aborted_op(tx, &op_manager, &gateways)
                                                        .await
                                                {
                                                    tracing::warn!(
                                                        tx = %tx,
                                                        error = ?err,
                                                        phase = "error",
                                                        "Failed to propagate aborted operation"
                                                    );
                                                }
                                            }
                                            Ok(None) => {
                                                tracing::error!(
                                                    tx = %tx,
                                                    peer_addr = %target_addr,
                                                    phase = "error",
                                                    "Response channel closed before connection result received"
                                                );
                                                if let Err(err) =
                                                    handle_aborted_op(tx, &op_manager, &gateways)
                                                        .await
                                                {
                                                    tracing::warn!(
                                                        tx = %tx,
                                                        error = ?err,
                                                        phase = "error",
                                                        "Failed to propagate aborted operation"
                                                    );
                                                }
                                            }
                                            Err(_) => {
                                                tracing::error!(
                                                    tx = %tx,
                                                    peer_addr = %target_addr,
                                                    phase = "timeout",
                                                    "Timeout waiting for connection establishment"
                                                );
                                                if let Err(err) =
                                                    handle_aborted_op(tx, &op_manager, &gateways)
                                                        .await
                                                {
                                                    tracing::warn!(
                                                        tx = %tx,
                                                        error = ?err,
                                                        phase = "error",
                                                        "Failed to propagate aborted operation"
                                                    );
                                                }
                                            }
                                        }
                                    });
                                }
                            }
                        }
                        ConnEvent::TransportClosed {
                            remote_addr,
                            error,
                            connection_id,
                        } => {
                            tracing::debug!(
                                peer_addr = %remote_addr,
                                error = ?error,
                                connection_id,
                                phase = "disconnect",
                                "Transport connection closed"
                            );
                        }
                        ConnEvent::StreamSend {
                            target_addr,
                            stream_id,
                            data,
                            metadata,
                            completion_tx,
                        } => {
                            // Route stream data to the per-connection channel.
                            // If dispatch fails, signal completion_tx so the
                            // broadcast queue releases its semaphore permit
                            // immediately instead of waiting for the 120s timeout.
                            if let Some(peer_connection) = ctx.connections.get(&target_addr) {
                                if let Err(e) = peer_connection
                                    .sender
                                    .send(Right(ConnEvent::StreamSend {
                                        target_addr,
                                        stream_id,
                                        data,
                                        metadata,
                                        completion_tx,
                                    }))
                                    .await
                                {
                                    tracing::error!(
                                        stream_id = %stream_id,
                                        peer_addr = %target_addr,
                                        error = %e,
                                        phase = "error",
                                        "Failed to send stream data to peer connection"
                                    );
                                    // The completion_tx is inside the failed send;
                                    // extract and signal it so the permit is released.
                                    if let Right(ConnEvent::StreamSend {
                                        completion_tx: Some(tx),
                                        ..
                                    }) = e.0
                                    {
                                        let _ignored = tx.send(());
                                    }
                                }
                            } else {
                                tracing::warn!(
                                    stream_id = %stream_id,
                                    peer_addr = %target_addr,
                                    phase = "error",
                                    "No connection found for stream send target"
                                );
                                // Signal completion so the broadcast queue permit
                                // is released immediately.
                                if let Some(tx) = completion_tx {
                                    let _ignored = tx.send(());
                                }
                            }
                        }
                        ConnEvent::PipeStream {
                            target_addr,
                            outbound_stream_id,
                            inbound_handle,
                            metadata,
                        } => {
                            // Route piped stream to the per-connection channel
                            if let Some(peer_connection) = ctx.connections.get(&target_addr) {
                                if let Err(e) = peer_connection
                                    .sender
                                    .send(Right(ConnEvent::PipeStream {
                                        target_addr,
                                        outbound_stream_id,
                                        inbound_handle,
                                        metadata,
                                    }))
                                    .await
                                {
                                    tracing::error!(
                                        stream_id = %outbound_stream_id,
                                        peer_addr = %target_addr,
                                        error = %e,
                                        phase = "error",
                                        "Failed to pipe stream to peer connection"
                                    );
                                }
                            } else {
                                tracing::warn!(
                                    stream_id = %outbound_stream_id,
                                    peer_addr = %target_addr,
                                    phase = "error",
                                    "No connection found for pipe stream target"
                                );
                            }
                        }
                        ConnEvent::ClosedChannel(reason) => {
                            match reason {
                                ChannelCloseReason::Bridge
                                | ChannelCloseReason::Controller
                                | ChannelCloseReason::Notification
                                | ChannelCloseReason::OpExecution => {
                                    // All ClosedChannel events are critical - the transport is unable to establish
                                    // more connections, rendering this peer useless. Perform cleanup and shutdown.
                                    let is_gateway = ctx.bridge.op_manager.ring.is_gateway();
                                    let active_conns = ctx.connections.len();
                                    // Write to stderr directly to ensure this survives process exit
                                    // (tracing output may be buffered and lost during shutdown)
                                    eprintln!(
                                        "CRITICAL: Channel closed reason={reason:?} is_gateway={is_gateway} active_connections={active_conns} - shutting down"
                                    );
                                    tracing::error!(
                                        reason = ?reason,
                                        is_gateway,
                                        active_connections = active_conns,
                                        phase = "shutdown",
                                        "Critical channel closed - performing cleanup and shutting down"
                                    );

                                    // Clean up all active connections
                                    let peers_to_cleanup: Vec<_> = ctx
                                        .connections
                                        .iter()
                                        .map(|(addr, entry)| (*addr, entry.pub_key.clone()))
                                        .collect();
                                    for (peer_addr, pub_key_opt) in peers_to_cleanup {
                                        tracing::debug!(
                                            peer_addr = %peer_addr,
                                            phase = "cleanup",
                                            "Cleaning up connection due to channel closure"
                                        );

                                        // Clean up ring state - construct PeerKeyLocation with pub_key if available
                                        let peer = if let Some(pub_key) = pub_key_opt.clone() {
                                            PeerKeyLocation::new(pub_key, peer_addr)
                                        } else {
                                            // Use our own pub_key as placeholder if we don't know the peer's
                                            PeerKeyLocation::new(
                                                (*ctx
                                                    .bridge
                                                    .op_manager
                                                    .ring
                                                    .connection_manager
                                                    .pub_key)
                                                    .clone(),
                                                peer_addr,
                                            )
                                        };
                                        let prune_result = ctx
                                            .bridge
                                            .op_manager
                                            .ring
                                            .prune_connection(PeerId::new(
                                                peer.pub_key().clone(),
                                                peer_addr,
                                            ))
                                            .await;

                                        // Note: In the simplified architecture (2026-01), subscriptions are lease-based
                                        // and don't require explicit pruning notifications. Just handle orphaned transactions.

                                        // Handle orphaned transactions immediately (retry via alternate routes)
                                        ctx.bridge
                                            .handle_orphaned_transactions(
                                                prune_result.orphaned_transactions,
                                                &ctx.gateways,
                                            )
                                            .await;

                                        if prune_result.became_unready {
                                            ctx.handle_broadcast_ready_state(false).await;
                                        }

                                        if let Some(ref pub_key) = pub_key_opt {
                                            ctx.bridge.op_manager.on_ring_connection_lost(pub_key);
                                        }

                                        // Remove from connection map
                                        tracing::trace!(
                                            peer_addr = %peer_addr,
                                            conn_map_size = ctx.connections.len(),
                                            reason = "channel_closed",
                                            "Removing connection from tracking map"
                                        );
                                        ctx.connections.remove(&peer_addr);
                                        if let Some(pub_key) = pub_key_opt {
                                            ctx.addr_by_pub_key.remove(&pub_key);
                                        }

                                        // Notify handshake handler to clean up
                                        if let Err(error) = handshake_cmd_sender
                                            .send(HandshakeCommand::DropConnection {
                                                peer: peer.clone(),
                                            })
                                            .await
                                        {
                                            tracing::warn!(
                                                peer = %peer,
                                                error = ?error,
                                                phase = "cleanup",
                                                "Failed to send drop connection command during cleanup"
                                            );
                                        }
                                    }

                                    // Clean up reservations for in-progress connections
                                    // These are connections that started handshake but haven't completed yet
                                    // Notifying the callbacks will trigger the calling code to clean up reservations
                                    tracing::debug!(
                                        awaiting_count = state.awaiting_connection.len(),
                                        phase = "cleanup",
                                        "Cleaning up in-progress connection reservations"
                                    );

                                    for (addr, mut callbacks) in state.awaiting_connection.drain() {
                                        tracing::debug!(
                                            peer_addr = %addr,
                                            callbacks = callbacks.len(),
                                            phase = "cleanup",
                                            "Notifying awaiting connection of shutdown"
                                        );
                                        // Best effort notification during shutdown - receiver may already be dropped
                                        for mut callback in callbacks.drain(..) {
                                            #[allow(clippy::let_underscore_must_use)]
                                            let _ = callback.send_result(Err(())).await;
                                        }
                                    }

                                    // Clean up pending operation result callbacks
                                    // Drop all waiting senders so callers see channel closure
                                    let pending_count = state.pending_op_results.len();
                                    if pending_count > 0 {
                                        tracing::debug!(
                                            pending_count,
                                            phase = "cleanup",
                                            "Draining pending_op_results"
                                        );
                                        for _ in 0..pending_count {
                                            crate::config::GlobalTestMetrics::record_pending_op_remove();
                                        }
                                        state.pending_op_results.drain();
                                    }

                                    tracing::info!(
                                        phase = "shutdown",
                                        "Cleanup complete - exiting event loop"
                                    );
                                    graceful_shutdown = true;
                                    break;
                                }
                            }
                        }
                        ConnEvent::NodeAction(action) => match action {
                            NodeEvent::DropConnection(peer_addr) => {
                                if !ctx
                                    .drop_connection_by_addr(peer_addr, &handshake_cmd_sender)
                                    .await
                                {
                                    tracing::debug!(
                                        peer_addr = %peer_addr,
                                        phase = "disconnect",
                                        "Drop connection requested for unknown address - ignoring"
                                    );
                                }
                            }
                            NodeEvent::DropAllConnections => {
                                let all_addrs: Vec<SocketAddr> =
                                    ctx.connections.keys().copied().collect();
                                tracing::warn!(
                                    count = all_addrs.len(),
                                    "DropAllConnections: closing all connections (suspend/resume recovery)"
                                );
                                // Use drop_connection_by_addr (blocking, 1s timeout) here
                                // rather than drop_zombie_connection because these may be
                                // healthy connections that need proper close notification.
                                // The deadlock risk is low: DropAllConnections fires only
                                // during suspend/resume recovery (rare), and during resume
                                // there are few inbound connections to fill the events channel.
                                for peer_addr in all_addrs {
                                    ctx.drop_connection_by_addr(peer_addr, &handshake_cmd_sender)
                                        .await;
                                }
                            }
                            NodeEvent::ConnectPeer {
                                peer,
                                tx,
                                callback,
                                is_gw: transient,
                            } => {
                                tracing::debug!(
                                    tx = %tx,
                                    peer = %peer,
                                    peer_addr = ?peer.socket_addr(),
                                    transient,
                                    phase = "connect",
                                    "Received connect peer request"
                                );
                                ctx.handle_connect_peer(
                                    peer,
                                    Box::new(callback),
                                    tx,
                                    &handshake_cmd_sender,
                                    &mut state,
                                    transient,
                                )
                                .await?;
                            }
                            NodeEvent::ExpectPeerConnection { addr } => {
                                tracing::debug!(
                                    peer_addr = %addr,
                                    direction = "inbound",
                                    phase = "connect",
                                    "Expecting peer connection - registering with handshake handler"
                                );
                                state.expected_inbound.expect_incoming(addr);
                                // We don't know the peer's public key yet for expected inbound connections,
                                // so use our own pub_key as a placeholder. The handshake will update it.
                                let peer_placeholder = PeerKeyLocation::new(
                                    (*ctx.bridge.op_manager.ring.connection_manager.pub_key)
                                        .clone(),
                                    addr,
                                );
                                if let Err(error) = handshake_cmd_sender
                                    .send(HandshakeCommand::ExpectInbound {
                                        peer: peer_placeholder,
                                        transaction: None,
                                        transient: false,
                                    })
                                    .await
                                {
                                    tracing::warn!(
                                        peer_addr = %addr,
                                        error = ?error,
                                        phase = "connect",
                                        "Failed to enqueue expect inbound command - connection may be rejected"
                                    );
                                }
                            }
                            NodeEvent::QueryConnections { callback } => {
                                // Reconstruct PeerKeyLocations from stored connections
                                let total_connections = ctx.connections.len();
                                let connections: Vec<PeerKeyLocation> = ctx
                                    .connections
                                    .iter()
                                    .filter_map(|(addr, entry)| {
                                        // Only include connections where we know the public key
                                        entry.pub_key.as_ref().map(|pub_key| {
                                            PeerKeyLocation::new(pub_key.clone(), *addr)
                                        })
                                    })
                                    .collect();
                                let with_pub_key = connections.len();
                                tracing::debug!(
                                    total_connections,
                                    with_pub_key,
                                    phase = "query",
                                    "Returning query connections result"
                                );
                                match timeout(
                                    Duration::from_secs(1),
                                    callback.send(QueryResult::Connections(connections)),
                                )
                                .await
                                {
                                    Ok(Ok(())) => {}
                                    Ok(Err(send_error)) => {
                                        tracing::error!(
                                            error = ?send_error,
                                            phase = "error",
                                            "Failed to send connections query result"
                                        );
                                    }
                                    Err(elapsed) => {
                                        tracing::error!(
                                            error = ?elapsed,
                                            phase = "timeout",
                                            "Timeout sending connections query result"
                                        );
                                    }
                                }
                            }
                            NodeEvent::QuerySubscriptions { callback } => {
                                // Get network subscriptions from OpManager
                                let network_subs = op_manager.get_network_subscriptions();
                                // Convert PeerKeyLocations to SocketAddrs for NetworkDebugInfo
                                let network_subs: Vec<(
                                    freenet_stdlib::prelude::ContractKey,
                                    Vec<SocketAddr>,
                                )> = network_subs
                                    .into_iter()
                                    .map(|(key, peers)| {
                                        let addrs = peers
                                            .into_iter()
                                            .filter_map(|p| p.socket_addr())
                                            .collect();
                                        (key, addrs)
                                    })
                                    .collect();

                                // Get application subscriptions from contract executor
                                // For now, we'll send a query to the contract handler
                                let (tx, mut rx) = tokio::sync::mpsc::channel(1);

                                op_manager
                                    .notify_contract_handler(
                                        ContractHandlerEvent::QuerySubscriptions { callback: tx },
                                    )
                                    .await?;

                                let app_subscriptions =
                                    match timeout(Duration::from_secs(1), rx.recv()).await {
                                        Ok(Some(QueryResult::NetworkDebug(info))) => {
                                            info.application_subscriptions
                                        }
                                        _ => Vec::new(),
                                    };

                                // Log network subscription details for debugging
                                for (contract_key, peers) in &network_subs {
                                    if !peers.is_empty() {
                                        tracing::trace!(
                                            contract = %contract_key,
                                            subscriber_count = peers.len(),
                                            phase = "query",
                                            "Network subscription found"
                                        );
                                    }
                                }

                                // Reconstruct PeerKeyLocations from stored connections
                                let connections: Vec<PeerKeyLocation> = ctx
                                    .connections
                                    .iter()
                                    .filter_map(|(addr, entry)| {
                                        // Only include connections where we know the public key
                                        entry.pub_key.as_ref().map(|pub_key| {
                                            PeerKeyLocation::new(pub_key.clone(), *addr)
                                        })
                                    })
                                    .collect();
                                let debug_info = crate::message::NetworkDebugInfo {
                                    application_subscriptions: app_subscriptions,
                                    network_subscriptions: network_subs,
                                    connected_peers: connections,
                                };

                                match timeout(
                                    Duration::from_secs(1),
                                    callback.send(QueryResult::NetworkDebug(debug_info)),
                                )
                                .await
                                {
                                    Ok(Ok(())) => {}
                                    Ok(Err(send_error)) => {
                                        tracing::error!(
                                            error = ?send_error,
                                            phase = "error",
                                            "Failed to send subscriptions query result"
                                        );
                                    }
                                    Err(elapsed) => {
                                        tracing::error!(
                                            error = ?elapsed,
                                            phase = "timeout",
                                            "Timeout sending subscriptions query result"
                                        );
                                    }
                                }
                            }
                            NodeEvent::QueryNodeDiagnostics { config, callback } => {
                                use freenet_stdlib::client_api::{
                                    ContractState, NetworkInfo, NodeDiagnosticsResponse, NodeInfo,
                                    SystemMetrics,
                                };
                                use std::collections::HashMap;

                                let mut response = NodeDiagnosticsResponse {
                                    node_info: None,
                                    network_info: None,
                                    subscriptions: Vec::new(),
                                    contract_states: HashMap::new(),
                                    system_metrics: None,
                                    connected_peers_detailed: Vec::new(),
                                };

                                // Collect node information
                                if config.include_node_info {
                                    // Get stored location (set by ObservedAddress handler) and address
                                    // IMPORTANT: Use get_stored_location() rather than computing from
                                    // get_own_addr() because the address may be a fallback value for
                                    // peers behind NAT, while the stored location comes from the
                                    // externally observed address.
                                    let addr = op_manager.ring.connection_manager.get_own_addr();
                                    let location =
                                        op_manager.ring.connection_manager.get_stored_location();

                                    // Always include basic node info, but only include address/location if available
                                    response.node_info = Some(NodeInfo {
                                        peer_id: ctx.key_pair.public().to_string(),
                                        is_gateway: self.is_gateway,
                                        location: location.map(|loc| format!("{:.6}", loc.0)),
                                        listening_address: addr
                                            .map(|peer_addr| peer_addr.to_string()),
                                        uptime_seconds: 0, // TODO: implement actual uptime tracking
                                    });
                                }

                                // Collect network information
                                if config.include_network_info {
                                    let cm = &op_manager.ring.connection_manager;
                                    let connections_by_loc = cm.get_connections_by_location();
                                    let mut connected_peers = Vec::new();
                                    for conns in connections_by_loc.values() {
                                        for conn in conns {
                                            connected_peers.push((
                                                conn.location.pub_key().to_string(),
                                                conn.location
                                                    .socket_addr()
                                                    .expect("connection should have address")
                                                    .to_string(),
                                            ));
                                        }
                                    }
                                    connected_peers.sort_by(|a, b| a.0.cmp(&b.0));
                                    connected_peers.dedup_by(|a, b| a.0 == b.0);

                                    response.network_info = Some(NetworkInfo {
                                        active_connections: connected_peers.len(),
                                        connected_peers,
                                    });
                                }

                                // Collect subscription information
                                if config.include_subscriptions {
                                    // Get network subscriptions from OpManager
                                    let _network_subs = op_manager.get_network_subscriptions();

                                    // Get application subscriptions from contract executor
                                    let (tx, mut rx) = tokio::sync::mpsc::channel(1);
                                    if op_manager
                                        .notify_contract_handler(
                                            ContractHandlerEvent::QuerySubscriptions {
                                                callback: tx,
                                            },
                                        )
                                        .await
                                        .is_ok()
                                    {
                                        let app_subscriptions = match timeout(
                                            Duration::from_secs(1),
                                            rx.recv(),
                                        )
                                        .await
                                        {
                                            Ok(Some(QueryResult::NetworkDebug(info))) => {
                                                info.application_subscriptions
                                            }
                                            _ => Vec::new(),
                                        };

                                        response.subscriptions = app_subscriptions
                                            .into_iter()
                                            .map(|sub| {
                                                freenet_stdlib::client_api::SubscriptionInfo {
                                                    contract_key: sub.instance_id,
                                                    client_id: sub.client_id.into(),
                                                }
                                            })
                                            .collect();
                                    }
                                }

                                // Collect contract states.
                                // When contract_keys is empty, return ALL hosting contracts.
                                // When specific keys are provided, return only those.
                                // Note: subscriber_peer_ids is always empty because we use
                                // lease-based subscriptions rather than explicit subscriber tracking.
                                if config.contract_keys.is_empty() {
                                    let hosting_contracts = op_manager.ring.hosting_contract_keys();
                                    for contract_key in hosting_contracts {
                                        let is_subscribed =
                                            op_manager.ring.is_subscribed(&contract_key);
                                        let subscriber_count = if is_subscribed { 1 } else { 0 };
                                        response.contract_states.insert(
                                            contract_key,
                                            ContractState {
                                                subscribers: subscriber_count as u32,
                                                subscriber_peer_ids: Vec::new(),
                                                size_bytes: op_manager
                                                    .ring
                                                    .hosting_contract_size(&contract_key),
                                            },
                                        );
                                    }
                                } else {
                                    for contract_key in &config.contract_keys {
                                        let is_subscribed =
                                            op_manager.ring.is_subscribed(contract_key);
                                        let subscriber_count = if is_subscribed { 1 } else { 0 };
                                        response.contract_states.insert(
                                            *contract_key,
                                            ContractState {
                                                subscribers: subscriber_count as u32,
                                                subscriber_peer_ids: Vec::new(),
                                                size_bytes: op_manager
                                                    .ring
                                                    .hosting_contract_size(contract_key),
                                            },
                                        );
                                    }
                                }

                                // Collect topology-backed connection info (exclude transient transports).
                                let cm = &op_manager.ring.connection_manager;
                                let connections_by_loc = cm.get_connections_by_location();
                                let mut connected_peer_ids = Vec::new();
                                if config.include_detailed_peer_info {
                                    use freenet_stdlib::client_api::ConnectedPeerInfo;
                                    for conns in connections_by_loc.values() {
                                        for conn in conns {
                                            connected_peer_ids
                                                .push(conn.location.pub_key().to_string());
                                            response.connected_peers_detailed.push(
                                                ConnectedPeerInfo {
                                                    peer_id: conn.location.pub_key().to_string(),
                                                    address: conn
                                                        .location
                                                        .socket_addr()
                                                        .expect("connection should have address")
                                                        .to_string(),
                                                },
                                            );
                                        }
                                    }
                                } else {
                                    for conns in connections_by_loc.values() {
                                        connected_peer_ids.extend(
                                            conns.iter().map(|c| c.location.pub_key().to_string()),
                                        );
                                    }
                                }
                                connected_peer_ids.sort();
                                connected_peer_ids.dedup();

                                // Collect system metrics
                                if config.include_system_metrics {
                                    // Use the actual hosting cache count, not the subscriber count.
                                    // The hosting cache tracks contracts this node is actively hosting,
                                    // while subscribers tracks remote peers subscribed to updates.
                                    let hosting_contracts =
                                        op_manager.ring.hosting_contracts_count() as u32;
                                    // Log memory/operation metrics for debugging (#2928)
                                    let pending_ops = op_manager.pending_op_counts();
                                    let contract_waiters_count =
                                        op_manager.contract_waiters_count();
                                    let memory_rss_bytes = get_rss_bytes();
                                    tracing::info!(
                                        pending_connect = pending_ops[0],
                                        pending_put = pending_ops[1],
                                        pending_get = pending_ops[2],
                                        pending_subscribe = pending_ops[3],
                                        pending_update = pending_ops[4],
                                        contract_waiters = contract_waiters_count,
                                        memory_rss_bytes = ?memory_rss_bytes,
                                        "Node diagnostics: operation & memory metrics"
                                    );
                                    response.system_metrics = Some(SystemMetrics {
                                        active_connections: connected_peer_ids.len() as u32,
                                        hosting_contracts,
                                    });
                                }

                                match timeout(
                                    Duration::from_secs(2),
                                    callback.send(QueryResult::NodeDiagnostics(response)),
                                )
                                .await
                                {
                                    Ok(Ok(())) => {}
                                    Ok(Err(send_error)) => {
                                        tracing::error!(
                                            error = ?send_error,
                                            phase = "error",
                                            "Failed to send node diagnostics query result"
                                        );
                                    }
                                    Err(elapsed) => {
                                        tracing::error!(
                                            error = ?elapsed,
                                            phase = "timeout",
                                            "Timeout sending node diagnostics query result"
                                        );
                                    }
                                }
                            }
                            NodeEvent::TransactionTimedOut(tx) => {
                                // Clean up client subscription to prevent memory leak
                                // Clients are not notified - transactions simply expire silently
                                if let Some(clients) = state.tx_to_client.remove(&tx) {
                                    tracing::debug!(
                                        tx = %tx,
                                        client_count = clients.len(),
                                        phase = "timeout",
                                        "Cleaned up client subscriptions for timed out transaction"
                                    );
                                }
                                // Clean up executor callback sender to prevent unbounded
                                // HashMap growth (entries were inserted by handle_op_execution
                                // but never removed — see #2941)
                                if state.pending_op_results.remove(&tx).is_some() {
                                    crate::config::GlobalTestMetrics::record_pending_op_remove();
                                }
                            }
                            NodeEvent::TransactionCompleted(tx) => {
                                // Clean up client subscription after successful completion
                                state.tx_to_client.remove(&tx);
                                // Clean up executor callback sender
                                if state.pending_op_results.remove(&tx).is_some() {
                                    crate::config::GlobalTestMetrics::record_pending_op_remove();
                                }
                            }
                            NodeEvent::LocalSubscribeComplete {
                                tx,
                                key,
                                subscribed,
                                is_renewal,
                            } => {
                                // This event is only fired for STANDALONE subscriptions (no remote peers).
                                // Normal subscribe flow now goes through handle_op_result which sends
                                // results via result_router_tx directly.
                                tracing::debug!(
                                    tx = %tx,
                                    contract = %key,
                                    is_renewal,
                                    phase = "complete",
                                    "Standalone subscribe operation completed"
                                );

                                // If this is a child operation (e.g., Subscribe spawned by PUT),
                                // just mark it complete - parent operation handles client response.
                                if op_manager.is_sub_operation(tx) {
                                    tracing::debug!(
                                        tx = %tx,
                                        contract = %key,
                                        phase = "complete",
                                        "Completing standalone child subscribe operation"
                                    );
                                    op_manager.completed(tx);
                                    continue;
                                }

                                // Subscription renewals are node-internal operations with no
                                // client waiting for results. Skip client notification. See #2891.
                                if is_renewal {
                                    tracing::debug!(
                                        tx = %tx,
                                        contract = %key,
                                        phase = "complete",
                                        "Skipping client notification for standalone subscription renewal"
                                    );
                                    op_manager.completed(tx);
                                    continue;
                                }

                                // Standalone parent operation - send response to client
                                let response = Ok(HostResponse::ContractResponse(
                                    ContractResponse::SubscribeResponse { key, subscribed },
                                ));

                                // Use try_send to avoid blocking the event loop.
                                // If the result router channel is full, the client
                                // will see a timeout rather than deadlocking the node.
                                match op_manager.result_router_tx.try_send((tx, response)) {
                                    Ok(()) => {
                                        tracing::debug!(
                                            tx = %tx,
                                            phase = "response",
                                            "Sent standalone subscribe response to client"
                                        );
                                        if let Some(clients) = state.tx_to_client.remove(&tx) {
                                            tracing::trace!(
                                                tx = %tx,
                                                client_count = clients.len(),
                                                "Removed waiting clients for completed transaction"
                                            );
                                        } else if let Some(pos) = state
                                            .client_waiting_transaction
                                            .iter()
                                            .position(|(waiting, _)| match waiting {
                                                WaitingTransaction::Subscription {
                                                    contract_key,
                                                } => contract_key == key.id(),
                                                _ => false,
                                            })
                                        {
                                            let (_, clients) =
                                                state.client_waiting_transaction.remove(pos);
                                            tracing::trace!(
                                                tx = %tx,
                                                contract = %key,
                                                waiter_count = clients.len(),
                                                "Matched subscription waiters by contract"
                                            );
                                        } else {
                                            tracing::warn!(
                                                tx = %tx,
                                                phase = "complete",
                                                "Standalone subscribe complete but no waiting clients found"
                                            );
                                        }
                                    }
                                    Err(e) => {
                                        tracing::error!(
                                            tx = %tx,
                                            error = %e,
                                            phase = "error",
                                            "Failed to send standalone subscribe response to client \
                                             (result router channel full or closed)"
                                        );
                                    }
                                }
                            }
                            NodeEvent::BroadcastHostingUpdate { message } => {
                                ctx.handle_hosting_broadcast(message).await;
                            }
                            NodeEvent::BroadcastChangeInterests { added, removed } => {
                                ctx.handle_broadcast_change_interests(added, removed).await;
                            }
                            NodeEvent::SendInterestMessage { target, message } => {
                                ctx.handle_send_interest_message(target, message).await;
                            }
                            NodeEvent::Disconnect { cause } => {
                                tracing::info!(
                                    cause = ?cause,
                                    phase = "shutdown",
                                    "Disconnecting from network"
                                );
                                graceful_shutdown = true;
                                break;
                            }
                            NodeEvent::BroadcastStateChange { key, new_state } => {
                                ctx.handle_broadcast_state_change(&op_manager, key, new_state)
                                    .await;
                            }
                            NodeEvent::SyncStateToPeer {
                                key,
                                new_state,
                                target,
                            } => {
                                ctx.handle_sync_state_to_peer(&op_manager, key, new_state, target)
                                    .await;
                            }
                        },
                    }
                }
            }
        }
        if graceful_shutdown {
            // Return typed error for graceful shutdown - callers can use downcast_ref()
            // to identify this as a clean exit rather than an actual error
            Err(EventLoopExitReason::GracefulShutdown.into())
        } else {
            Err(EventLoopExitReason::UnexpectedStreamEnd.into())
        }
    }

    /// Drop a single connection by socket address: notifies the handshake handler,
    /// prunes topology, handles orphaned transactions, cleans up subscriptions,
    /// and forwards a DropConnection event to the per-peer listener.
    ///
    /// Returns `true` if the connection existed and was dropped.
    async fn drop_connection_by_addr(
        &mut self,
        peer_addr: SocketAddr,
        handshake_cmd_sender: &HandshakeCommandSender,
    ) -> bool {
        let Some(entry) = self.connections.get(&peer_addr) else {
            return false;
        };

        let peer = if let Some(ref pub_key) = entry.pub_key {
            PeerKeyLocation::new(pub_key.clone(), peer_addr)
        } else {
            PeerKeyLocation::new(
                (*self.bridge.op_manager.ring.connection_manager.pub_key).clone(),
                peer_addr,
            )
        };
        let pub_key_to_remove = entry.pub_key.clone();

        tracing::trace!(
            peer_addr = %peer_addr,
            conn_map_size = self.connections.len(),
            reason = "drop_requested",
            "Removing connection from tracking map"
        );

        if let Err(error) = handshake_cmd_sender
            .send(HandshakeCommand::DropConnection { peer: peer.clone() })
            .await
        {
            tracing::warn!(
                peer = %peer,
                error = ?error,
                phase = "disconnect",
                "Failed to enqueue drop connection command"
            );
        }

        // Immediately prune topology counters so we don't leak open connection slots.
        let prune_result = self
            .bridge
            .op_manager
            .ring
            .prune_connection(PeerId::new(peer.pub_key().clone(), peer_addr))
            .await;

        // Handle orphaned transactions immediately (retry via alternate routes)
        self.bridge
            .handle_orphaned_transactions(prune_result.orphaned_transactions, &self.gateways)
            .await;

        // Broadcast unready if we dropped below the readiness threshold
        if prune_result.became_unready {
            self.handle_broadcast_ready_state(false).await;
        }

        // Forget this peer's subscriptions and cached contract state
        self.bridge
            .op_manager
            .on_ring_connection_lost(peer.pub_key());

        if let Some(conn) = self.connections.remove(&peer_addr) {
            if let Some(pub_key) = pub_key_to_remove {
                self.addr_by_pub_key.remove(&pub_key);
            }
            // TODO: review: this could potentially leave garbage tasks in the background with peer listener
            match timeout(
                Duration::from_secs(1),
                conn.sender
                    .send(Right(ConnEvent::NodeAction(NodeEvent::DropConnection(
                        peer_addr,
                    )))),
            )
            .await
            {
                Ok(Ok(())) => {}
                Ok(Err(send_error)) => {
                    tracing::error!(
                        peer_addr = %peer_addr,
                        error = ?send_error,
                        phase = "error",
                        "Failed to send drop connection message to peer"
                    );
                }
                Err(elapsed) => {
                    tracing::error!(
                        peer_addr = %peer_addr,
                        error = ?elapsed,
                        phase = "timeout",
                        "Timeout while sending drop connection message"
                    );
                }
            }
        }

        true
    }

    /// Drop a zombie transport connection using non-blocking sends to the handshake
    /// driver. This prevents a deadlock where the event loop blocks sending
    /// DropConnection commands while the handshake driver blocks sending
    /// InboundConnection events back to the event loop (#3519).
    ///
    /// Unlike `drop_connection_by_addr`, this method:
    /// - Uses `try_send` for the handshake command (skips if channel full)
    /// - Uses a shorter timeout for the per-connection drop notification
    async fn drop_zombie_connection(
        &mut self,
        peer_addr: SocketAddr,
        handshake_cmd_sender: &HandshakeCommandSender,
    ) {
        let Some(entry) = self.connections.get(&peer_addr) else {
            return;
        };

        let peer = if let Some(ref pub_key) = entry.pub_key {
            PeerKeyLocation::new(pub_key.clone(), peer_addr)
        } else {
            PeerKeyLocation::new(
                (*self.bridge.op_manager.ring.connection_manager.pub_key).clone(),
                peer_addr,
            )
        };
        let pub_key_to_remove = entry.pub_key.clone();

        // Non-blocking send to handshake driver to avoid deadlock (#3519).
        // For zombie transports (already established connections), the expected_inbound
        // entry was already consumed when the inbound connection arrived, so the
        // DropConnection command is effectively a no-op.
        if !handshake_cmd_sender.try_send(HandshakeCommand::DropConnection { peer: peer.clone() }) {
            tracing::debug!(
                peer = %peer,
                "Handshake channel full during zombie cleanup, skipping handshake notification"
            );
        }

        // Prune from ring topology (non-blocking — only takes write locks, no cross-task channels)
        let prune_result = self
            .bridge
            .op_manager
            .ring
            .prune_connection(PeerId::new(peer.pub_key().clone(), peer_addr))
            .await;

        // Handle orphaned transactions
        self.bridge
            .handle_orphaned_transactions(prune_result.orphaned_transactions, &self.gateways)
            .await;

        if prune_result.became_unready {
            self.handle_broadcast_ready_state(false).await;
        }

        // Forget subscriptions and cached contract state
        self.bridge
            .op_manager
            .on_ring_connection_lost(peer.pub_key());

        // Remove from transport tracking and notify the per-connection task
        if let Some(conn) = self.connections.remove(&peer_addr) {
            if let Some(pub_key) = pub_key_to_remove {
                self.addr_by_pub_key.remove(&pub_key);
            }
            // Short timeout — zombie connections may already be dead
            match timeout(
                Duration::from_millis(100),
                conn.sender
                    .send(Right(ConnEvent::NodeAction(NodeEvent::DropConnection(
                        peer_addr,
                    )))),
            )
            .await
            {
                Ok(Ok(())) => {}
                Ok(Err(_)) | Err(_) => {
                    // Expected for zombie connections — the peer task may already be gone
                }
            }
        }
    }

    /// Process a SelectResult from the priority select stream
    async fn process_select_result(
        &mut self,
        result: priority_select::SelectResult,
        state: &mut EventListenerState,
        handshake_commands: &HandshakeCommandSender,
    ) -> anyhow::Result<EventResult> {
        let peer_id = &self.bridge.op_manager.ring.connection_manager.pub_key;

        use priority_select::SelectResult;
        match result {
            SelectResult::Notification(msg) => {
                tracing::debug!(
                    peer = %peer_id,
                    msg_present = msg.is_some(),
                    "PrioritySelect: notifications_receiver READY"
                );
                Ok(self.handle_notification_msg(msg))
            }
            SelectResult::OpExecution(msg) => {
                tracing::debug!(
                    peer = %peer_id,
                    "PrioritySelect: op_execution_receiver READY"
                );
                Ok(self.handle_op_execution(msg, state))
            }
            SelectResult::PeerConnection(msg) => {
                tracing::debug!(
                    peer = %peer_id,
                    "PrioritySelect: connection events READY"
                );
                self.handle_transport_event(msg, state, handshake_commands)
                    .await
            }
            SelectResult::ConnBridge(msg) => {
                tracing::debug!(
                    peer = %peer_id,
                    "PrioritySelect: conn_bridge_rx READY"
                );
                Ok(self.handle_bridge_msg(msg))
            }
            SelectResult::Handshake(result) => {
                tracing::debug!(
                    peer = %peer_id,
                        "PrioritySelect: handshake event READY"
                );
                match result {
                    Some(event) => {
                        self.handle_handshake_action(event, state, handshake_commands)
                            .await?;
                        Ok(EventResult::Continue)
                    }
                    None => {
                        tracing::warn!(
                            "Handshake handler stream closed; notifying pending callbacks"
                        );
                        self.handle_handshake_stream_closed(state).await?;
                        Ok(EventResult::Continue)
                    }
                }
            }
            SelectResult::NodeController(msg) => {
                tracing::debug!(
                    peer = %peer_id,
                    "PrioritySelect: node_controller READY"
                );
                Ok(self.handle_node_controller_msg(msg))
            }
            SelectResult::ClientTransaction(event_id) => {
                tracing::debug!(
                    peer = %peer_id,
                    "PrioritySelect: client_wait_for_transaction READY"
                );
                Ok(self.handle_client_transaction_subscription(event_id, state))
            }
            SelectResult::ExecutorTransaction(id) => {
                tracing::debug!(
                    peer = %peer_id,
                    "PrioritySelect: executor_listener READY"
                );
                Ok(self.handle_executor_transaction(id, state))
            }
        }
    }

    async fn handle_inbound_message(
        &self,
        msg: NetMessage,
        source_addr: Option<SocketAddr>,
        op_manager: &Arc<OpManager>,
        state: &mut EventListenerState,
    ) -> anyhow::Result<()> {
        let tx = *msg.id();
        tracing::debug!(
            %tx,
            tx_type = ?tx.transaction_type(),
            ?source_addr,
            "Handling inbound NetMessage at event loop"
        );
        match msg {
            NetMessage::V1(NetMessageV1::Aborted(tx)) => {
                handle_aborted_op(tx, op_manager, &self.gateways).await?;
            }
            msg => {
                self.process_message(msg, source_addr, op_manager, None, state)
                    .await;
            }
        }
        Ok(())
    }

    async fn process_message(
        &self,
        msg: NetMessage,
        source_addr: Option<SocketAddr>,
        op_manager: &Arc<OpManager>,
        executor_callback_opt: Option<ExecutorToEventLoopChannel<crate::contract::Callback>>,
        state: &mut EventListenerState,
    ) {
        tracing::debug!(
            tx = %msg.id(),
            tx_type = ?msg.id().transaction_type(),
            msg_type = %msg,
            ?source_addr,
            peer = ?op_manager.ring.connection_manager.get_own_addr(),
            "process_message called - processing network message"
        );

        // Only use the callback if this message was initiated by the executor
        let executor_callback_opt = if state.pending_from_executor.remove(msg.id()) {
            executor_callback_opt
        } else {
            None
        };

        let span = tracing::info_span!(
            "process_network_message",
            transaction = %msg.id(),
            tx_type = %msg.id().transaction_type()
        );

        let pending_op_result = state.pending_op_results.get(msg.id()).cloned();

        GlobalExecutor::spawn(
            process_message_decoupled(
                msg,
                source_addr,
                op_manager.clone(),
                self.bridge.clone(),
                self.event_listener.trait_clone(),
                executor_callback_opt,
                pending_op_result,
            )
            .instrument(span),
        );
    }

    /// Looks up a connection by public key using the reverse lookup map.
    /// Returns the socket address and connection entry if found.
    fn connection_entry_by_pub_key(
        &self,
        pub_key: &TransportPublicKey,
    ) -> Option<(SocketAddr, &ConnectionEntry)> {
        self.addr_by_pub_key
            .get(pub_key)
            .and_then(|addr| self.connections.get(addr).map(|entry| (*addr, entry)))
    }

    async fn handle_connect_peer(
        &mut self,
        peer: PeerKeyLocation,
        mut callback: Box<dyn ConnectResultSender>,
        tx: Transaction,
        handshake_commands: &HandshakeCommandSender,
        state: &mut EventListenerState,
        transient: bool,
    ) -> anyhow::Result<()> {
        // Periodic cleanup of expired backoff entries (every 60 seconds)
        const BACKOFF_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
        if state.last_backoff_cleanup.elapsed() > BACKOFF_CLEANUP_INTERVAL {
            state.peer_backoff.cleanup_expired();
            state.last_backoff_cleanup = Instant::now();
        }

        let mut peer = peer;
        let mut peer_addr = peer
            .socket_addr()
            .unwrap_or_else(|| SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0));

        // IMPORTANT: Always try pub_key lookup first, not just for unspecified addresses.
        // The peer's advertised address (from PeerKeyLocation) may differ from the actual
        // TCP connection's remote address stored in self.connections. For example:
        // - PeerKeyLocation may have the peer's listening port (e.g., 37791)
        // - self.connections is keyed by the actual TCP source port (ephemeral port)
        // By looking up via pub_key (populated when we receive the first message from this
        // peer), we can find the correct connection entry regardless of address mismatch.
        if let Some((existing_addr, _)) = self.connection_entry_by_pub_key(peer.pub_key()) {
            if existing_addr != peer_addr {
                tracing::debug!(
                    tx = %tx,
                    peer = %peer,
                    advertised_addr = %peer_addr,
                    actual_addr = %existing_addr,
                    transient,
                    phase = "connect",
                    "Using existing connection (advertised address differs from actual)"
                );
            }
            peer_addr = existing_addr;
            peer = PeerKeyLocation::new(peer.pub_key().clone(), existing_addr);
        } else if peer_addr.ip().is_unspecified() {
            tracing::debug!(
                tx = %tx,
                transient,
                "ConnectPeer received unspecified address without existing connection reference"
            );
        }

        tracing::debug!(
            tx = %tx,
            peer = %peer,
            peer_addr = %peer_addr,
            transient,
            phase = "connect",
            "Initiating connection to peer"
        );
        if let Some(blocked_addrs) = &self.blocked_addresses {
            if blocked_addrs.contains(&peer_addr) {
                tracing::info!(
                    tx = %tx,
                    remote = %peer_addr,
                    "Outgoing connection to peer blocked by local policy"
                );
                callback
                    .send_result(Err(()))
                    .await
                    .inspect_err(|error| {
                        tracing::debug!(
                            remote = %peer_addr,
                            ?error,
                            "Failed to notify caller about blocked connection"
                        );
                    })
                    .ok();
                return Ok(());
            }
            tracing::debug!(
                tx = %tx,
                "Blocked addresses: {:?}, peer addr: {}",
                blocked_addrs,
                peer_addr
            );
        }

        // If a transient transport already exists, promote it without dialing anew.
        if self.connections.contains_key(&peer_addr) {
            tracing::info!(
                tx = %tx,
                remote = %peer,
                transient,
                "connect_peer: reusing existing transport / promoting transient if present"
            );
            let connection_manager = &self.bridge.op_manager.ring.connection_manager;
            let was_transient = connection_manager.drop_transient(peer_addr);

            // Promote if: (a) was tracked as transient, OR (b) not already in ring.
            // Case (b) handles expired transient TTL — transport preserved per 76058cf4
            // but tracking entry gone. Without this, CONNECT succeeds but peer is never
            // added to ring topology (#3113).
            let needs_promotion =
                was_transient.is_some() || !connection_manager.is_in_ring(peer_addr);
            let transient_expired = was_transient.is_none() && needs_promotion;

            if needs_promotion {
                let loc = was_transient
                    .and_then(|e| e.location)
                    .unwrap_or_else(|| Location::from_address(&peer_addr));
                // Don't re-run should_accept() here — the connection was already
                // accepted at the protocol level (the CONNECT terminus handler in connect.rs) and NAT traversal
                // already succeeded. Re-evaluating the probabilistic Kleinberg filter
                // would discard hard-won connections for no capacity reason (#3545).
                // Only enforce the hard max_connections cap as a resource safety limit.
                let current = connection_manager.connection_count();
                if current >= connection_manager.max_connections {
                    tracing::warn!(
                        tx = %tx,
                        %peer,
                        current_connections = current,
                        max_connections = connection_manager.max_connections,
                        %loc,
                        transient_expired,
                        "connect_peer: rejecting transient promotion to enforce cap"
                    );
                    callback
                        .send_result(Err(()))
                        .await
                        .inspect_err(|err| {
                            tracing::debug!(
                                tx = %tx,
                                remote = %peer,
                                ?err,
                                "connect_peer: failed to notify cap-rejection callback"
                            );
                        })
                        .ok();
                    return Ok(());
                }
                let just_became_ready = self
                    .bridge
                    .op_manager
                    .ring
                    .add_connection(loc, PeerId::new(peer.pub_key().clone(), peer_addr), true)
                    .await;
                tracing::info!(
                    tx = %tx,
                    remote = %peer,
                    transient_expired,
                    "connect_peer: promoted to ring"
                );

                // Update the dashboard/network_status with the peer's location.
                // Without this, peers promoted via this path show "—" on the
                // dashboard because the initial record_peer_connected(addr, None, None)
                // from handle_successful_connection is never updated.
                let pkl = crate::ring::PeerKeyLocation::new(peer.pub_key().clone(), peer_addr);
                crate::node::network_status::record_peer_connected(
                    peer_addr,
                    Some(loc.as_f64()),
                    Some(pkl),
                );

                // tell the promoted peer about our subscriptions and cache
                for (target, msg) in self
                    .bridge
                    .op_manager
                    .on_ring_connection_established(peer_addr, peer.pub_key())
                {
                    if let Err(e) = self.bridge.send(target, msg).await {
                        tracing::warn!(
                            %peer_addr,
                            error = %e,
                            "Failed to send interest/cache data on transient promotion"
                        );
                    }
                }

                // Broadcast readiness if we just crossed the threshold
                if just_became_ready {
                    self.handle_broadcast_ready_state(true).await;
                }
            }

            // Now that we know the peer's identity (from ConnectRequest), update the
            // transport-level tracking so QueryConnections returns this peer.
            if let Some(entry) = self.connections.get_mut(&peer_addr) {
                if entry.pub_key.is_none() {
                    entry.pub_key = Some(peer.pub_key().clone());
                    self.addr_by_pub_key
                        .insert(peer.pub_key().clone(), peer_addr);
                    tracing::info!(
                        tx = %tx,
                        %peer_addr,
                        pub_key = %peer.pub_key(),
                        "connect_peer: updated transport entry with peer identity"
                    );
                } else {
                    tracing::info!(
                        tx = %tx,
                        %peer_addr,
                        existing_pub_key = ?entry.pub_key,
                        "connect_peer: transport entry already has pub_key"
                    );
                }
            } else {
                tracing::warn!(
                    tx = %tx,
                    %peer_addr,
                    "connect_peer: no transport entry found for peer_addr"
                );
            }

            // Return the remote peer's address we are connected to.
            let resolved_addr = peer
                .socket_addr()
                .expect("connected peer should have socket address");
            callback
                .send_result(Ok((resolved_addr, None)))
                .await
                .inspect_err(|err| {
                    tracing::debug!(
                        tx = %tx,
                        remote = %peer,
                        ?err,
                        "connect_peer: failed to notify existing-connection callback"
                    );
                })
                .ok();
            return Ok(());
        }

        // Check if this peer is in backoff due to previous connection failures.
        // This check is done AFTER the existing connection check above, so that:
        // 1. If we already have a connection (possibly from inbound), we reuse it
        // 2. Backoff only blocks NEW outbound connection attempts
        // NOTE: Backoff runs before the max_connections pre-flight check
        // to avoid unnecessary work when the peer is in backoff.
        if !peer_addr.ip().is_unspecified() && state.peer_backoff.is_in_backoff(peer_addr) {
            let remaining = state
                .peer_backoff
                .remaining_backoff(peer_addr)
                .unwrap_or(Duration::ZERO);
            tracing::debug!(
                tx = %tx,
                peer = %peer,
                peer_addr = %peer_addr,
                remaining_secs = remaining.as_secs(),
                transient,
                phase = "connect",
                "Skipping connection attempt - peer in backoff"
            );
            callback
                .send_result(Err(()))
                .await
                .inspect_err(|error| {
                    tracing::debug!(
                        remote = %peer_addr,
                        ?error,
                        "Failed to notify caller about backoff-delayed connection"
                    );
                })
                .ok();
            return Ok(());
        }

        // Pre-flight max_connections check: avoid expensive NAT hole-punching when
        // we're already at capacity. We don't re-run should_accept() here because
        // the connection was already accepted at the protocol level (the CONNECT terminus handler in connect.rs)
        // and re-rolling the probabilistic Kleinberg filter would discard connections
        // after costly NAT traversal for no capacity reason (#3545).
        if !transient && !peer_addr.ip().is_unspecified() {
            let connection_manager = &self.bridge.op_manager.ring.connection_manager;
            let current = connection_manager.connection_count();
            if current >= connection_manager.max_connections {
                tracing::info!(
                    tx = %tx,
                    remote = %peer_addr,
                    current_connections = current,
                    max_connections = connection_manager.max_connections,
                    "connect_peer: skipping connection — at max_connections"
                );
                callback
                    .send_result(Err(()))
                    .await
                    .inspect_err(|err| {
                        tracing::debug!(
                            tx = %tx,
                            remote = %peer_addr,
                            ?err,
                            "connect_peer: failed to notify max-connections-rejection callback"
                        );
                    })
                    .ok();
                return Ok(());
            }
        }

        match state.awaiting_connection.entry(peer_addr) {
            std::collections::hash_map::Entry::Occupied(mut callbacks) => {
                let txs_entry = state.awaiting_connection_txs.entry(peer_addr).or_default();
                if !txs_entry.contains(&tx) {
                    txs_entry.push(tx);
                }
                tracing::debug!(
                    tx = %tx,
                    remote = %peer_addr,
                    pending = callbacks.get().len(),
                    transient,
                    "Connection already pending, queuing additional requester"
                );
                callbacks.get_mut().push(callback);
                tracing::info!(
                    tx = %tx,
                    remote = %peer_addr,
                    pending = callbacks.get().len(),
                    pending_txs = ?txs_entry,
                    transient,
                    "connect_peer: connection already pending, queued callback"
                );
                return Ok(());
            }
            std::collections::hash_map::Entry::Vacant(entry) => {
                let txs_entry = state.awaiting_connection_txs.entry(peer_addr).or_default();
                txs_entry.push(tx);
                tracing::debug!(
                        tx = %tx,
                    remote = %peer_addr,
                    transient,
                    "connect_peer: registering new pending connection"
                );
                entry.insert(vec![callback]);
                tracing::info!(
                tx = %tx,
                remote = %peer_addr,
                pending = 1,
                    pending_txs = ?txs_entry,
                    transient,
                    "connect_peer: registered new pending connection"
                );
                state.expected_inbound.expect_incoming(peer_addr);
            }
        }

        if let Err(error) = handshake_commands
            .send(HandshakeCommand::Connect {
                peer: peer.clone(),
                transaction: tx,
                transient,
            })
            .await
        {
            tracing::warn!(
                tx = %tx,
                remote = %peer_addr,
                transient,
                ?error,
                "Failed to enqueue connect command"
            );
            self.bridge
                .op_manager
                .ring
                .connection_manager
                .prune_in_transit_connection(peer_addr);

            // Note: We intentionally do NOT record backoff here. This failure is a LOCAL
            // channel issue (handshake command queue full), not a failure to connect to the
            // REMOTE peer. Backoff should only apply to actual connection failures
            // (OutboundFailed), not local resource contention.

            let pending_txs = state.awaiting_connection_txs.remove(&peer_addr);
            if let Some(callbacks) = state.awaiting_connection.remove(&peer_addr) {
                tracing::debug!(
                    tx = %tx,
                    remote = %peer_addr,
                    callbacks = callbacks.len(),
                    transient,
                    "Cleaning up callbacks after connect command failure"
                );
                for mut cb in callbacks {
                    cb.send_result(Err(()))
                        .await
                        .inspect_err(|send_err| {
                            tracing::debug!(
                                remote = %peer_addr,
                                ?send_err,
                                "Failed to deliver connect command failure to awaiting callback"
                            );
                        })
                        .ok();
                }
            }
            if let Some(pending_txs) = pending_txs {
                tracing::debug!(
                    remote = %peer_addr,
                    pending_txs = ?pending_txs,
                    "Removed pending transactions after connect command failure"
                );
            }
        } else {
            tracing::debug!(
                tx = %tx,
                remote = %peer_addr,
                transient,
                "connect_peer: handshake command dispatched"
            );
        }

        Ok(())
    }

    async fn handle_handshake_action(
        &mut self,
        event: HandshakeEvent,
        state: &mut EventListenerState,
        handshake_commands: &HandshakeCommandSender,
    ) -> anyhow::Result<()> {
        tracing::info!(?event, "handle_handshake_action: received handshake event");
        match event {
            HandshakeEvent::InboundConnection {
                transaction,
                peer,
                connection,
                transient,
            } => {
                tracing::info!(provided = ?peer, transient, tx = ?transaction, "InboundConnection event");
                let _conn_manager = &self.bridge.op_manager.ring.connection_manager;
                let remote_addr = connection.remote_addr();

                if let Some(blocked_addrs) = &self.blocked_addresses {
                    if blocked_addrs.contains(&remote_addr) {
                        tracing::info!(
                            remote = %remote_addr,
                            transient,
                            transaction = ?transaction,
                            "Inbound connection blocked by local policy"
                        );
                        return Ok(());
                    }
                }

                // For transient inbound connections, peer may be None - we don't know the
                // peer's identity yet. The identity will be learned when the peer sends
                // its first message (e.g., ConnectRequest).
                if peer.is_none() {
                    tracing::info!(
                        remote = %remote_addr,
                        transient,
                        transaction = ?transaction,
                        "Inbound connection arrived without matching expectation; accepting provisionally"
                    );
                }

                tracing::info!(
                    remote = %remote_addr,
                    peer_known = peer.is_some(),
                    transient,
                    transaction = ?transaction,
                    "Inbound connection established"
                );

                // Clear backoff on inbound connection: the peer is demonstrably alive (#3252).
                if !remote_addr.ip().is_unspecified() {
                    state.peer_backoff.record_success(remote_addr);
                }

                // Treat only transient connections as transient. Normal inbound dials (including
                // gateway bootstrap from peers) should be promoted into the ring once established.
                let is_transient = transient;

                // Pass peer directly - it may be None for transient connections
                self.handle_successful_connection(
                    peer,
                    connection,
                    state,
                    None,
                    is_transient,
                    handshake_commands,
                )
                .await?;
            }
            HandshakeEvent::OutboundEstablished {
                transaction,
                peer,
                connection,
                transient,
            } => {
                // Outbound peers must have known addresses - use type-safe conversion
                // If conversion fails, log error but continue since connection is established
                let peer_addr = match KnownPeerKeyLocation::try_from(&peer) {
                    Ok(k) => k.socket_addr(),
                    Err(e) => {
                        tracing::error!(
                            transaction = %transaction,
                            pub_key = %e.pub_key,
                            "INTERNAL ERROR: outbound connection established but peer has unknown address"
                        );
                        // Use peer's pub_key in log, but we need an address for downstream logging
                        // This should never happen, so use unspecified as last resort
                        SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0)
                    }
                };
                tracing::info!(
                    remote = %peer_addr,
                    transient,
                    transaction = %transaction,
                    "Outbound connection established"
                );

                // Clear backoff and failed-addr tracking on successful connection
                if !peer_addr.ip().is_unspecified() {
                    state.peer_backoff.record_success(peer_addr);
                    self.bridge
                        .op_manager
                        .ring
                        .connection_manager
                        .clear_failed_addr(peer_addr);
                }

                // For outbound connections, respect the transient flag from the handshake.
                // Gateway connections should remain transient until CONNECT acceptance.
                self.handle_successful_connection(
                    Some(peer),
                    connection,
                    state,
                    None,
                    transient,
                    handshake_commands,
                )
                .await?;
            }
            HandshakeEvent::OutboundFailed {
                transaction,
                peer,
                error,
                transient,
            } => {
                // Outbound peers must have known addresses - extract once for reuse
                let peer_addr = match KnownPeerKeyLocation::try_from(&peer) {
                    Ok(k) => k.socket_addr(),
                    Err(e) => {
                        // This is an internal consistency error - we initiated this connection,
                        // so we should always know the target address
                        tracing::error!(
                            transaction = %transaction,
                            pub_key = %e.pub_key,
                            "INTERNAL ERROR: outbound connection failed but peer has unknown address"
                        );
                        return Ok(());
                    }
                };

                tracing::info!(
                    remote = %peer_addr,
                    transient,
                    transaction = %transaction,
                    ?error,
                    "Outbound connection failed"
                );

                self.bridge
                    .op_manager
                    .ring
                    .connection_manager
                    .prune_in_transit_connection(peer_addr);

                // Only record transport-level failures (NAT traversal, timeout, I/O)
                // so future CONNECT requests avoid routing to this peer. Protocol
                // errors or rejections don't indicate the peer is unreachable.
                if matches!(
                    error,
                    ConnectionError::TransportError(_)
                        | ConnectionError::Timeout
                        | ConnectionError::IOError(_)
                ) {
                    self.bridge
                        .op_manager
                        .ring
                        .connection_manager
                        .record_failed_addr(peer_addr);
                }

                // Record failure for the connecting page diagnostics
                let failure_reason = match &error {
                    ConnectionError::TransportError(msg) => {
                        crate::node::network_status::classify_transport_error(msg)
                    }
                    ConnectionError::Timeout => crate::node::network_status::FailureReason::Timeout,
                    ConnectionError::IOError(msg) => {
                        crate::node::network_status::FailureReason::Other(msg.clone())
                    }
                    other @ ConnectionError::LocationUnknown
                    | other @ ConnectionError::SendNotCompleted(_)
                    | other @ ConnectionError::UnexpectedReq
                    | other @ ConnectionError::Serialization(_)
                    | other @ ConnectionError::FailedConnectOp
                    | other @ ConnectionError::UnwantedConnection
                    | other @ ConnectionError::AddressBlocked(_) => {
                        crate::node::network_status::FailureReason::Other(other.to_string())
                    }
                };
                crate::node::network_status::record_gateway_failure(peer_addr, failure_reason);

                // Record failure for exponential backoff to prevent rapid retries
                if !peer_addr.ip().is_unspecified() {
                    state.peer_backoff.record_failure(peer_addr);
                }

                let pending_txs = state
                    .awaiting_connection_txs
                    .remove(&peer_addr)
                    .unwrap_or_default();

                if let Some(callbacks) = state.awaiting_connection.remove(&peer_addr) {
                    tracing::debug!(
                        remote = %peer_addr,
                        callbacks = callbacks.len(),
                        pending_txs = ?pending_txs,
                        transient,
                        "Notifying callbacks after outbound failure"
                    );

                    let mut callbacks = callbacks.into_iter();
                    if let Some(mut cb) = callbacks.next() {
                        cb.send_result(Err(()))
                            .await
                            .inspect_err(|err| {
                                tracing::debug!(
                                    remote = %peer_addr,
                                    ?err,
                                    "Failed to deliver outbound failure notification"
                                );
                            })
                            .ok();
                    }
                    for mut cb in callbacks {
                        cb.send_result(Err(()))
                            .await
                            .inspect_err(|err| {
                                tracing::debug!(
                                    remote = %peer_addr,
                                    ?err,
                                    "Failed to deliver secondary outbound failure notification"
                                );
                            })
                            .ok();
                    }
                }
            }
        }
        Ok(())
    }

    async fn handle_handshake_stream_closed(
        &mut self,
        state: &mut EventListenerState,
    ) -> anyhow::Result<()> {
        if state.awaiting_connection.is_empty() {
            return Ok(());
        }

        tracing::warn!(
            awaiting = state.awaiting_connection.len(),
            "Handshake driver closed; notifying pending callbacks"
        );

        let awaiting = std::mem::take(&mut state.awaiting_connection);
        let awaiting_txs = std::mem::take(&mut state.awaiting_connection_txs);

        for (addr, callbacks) in awaiting {
            let pending_txs = awaiting_txs.get(&addr).cloned().unwrap_or_default();
            tracing::debug!(
                remote = %addr,
                callbacks = callbacks.len(),
                pending_txs = ?pending_txs,
                "Delivering handshake driver shutdown notification"
            );
            for mut cb in callbacks {
                cb.send_result(Err(()))
                    .await
                    .inspect_err(|err| {
                        tracing::debug!(
                            remote = %addr,
                            ?err,
                            "Failed to deliver handshake driver shutdown notification"
                        );
                    })
                    .ok();
            }
        }

        Ok(())
    }

    async fn handle_successful_connection(
        &mut self,
        peer_id: Option<PeerKeyLocation>,
        connection: Box<dyn PeerConnectionApi>,
        state: &mut EventListenerState,
        remaining_checks: Option<usize>,
        is_transient: bool,
        handshake_commands: &HandshakeCommandSender,
    ) -> anyhow::Result<()> {
        let mut broadcast_ready: Option<bool> = None;
        let connection_manager = &self.bridge.op_manager.ring.connection_manager;
        // For transient connections, we may not know the peer's identity yet - use connection's remote_addr
        let peer_addr = peer_id
            .as_ref()
            .and_then(|p| p.socket_addr())
            .unwrap_or_else(|| connection.remote_addr());

        if is_transient && !connection_manager.try_register_transient(peer_addr, None) {
            tracing::warn!(
                remote = %peer_addr,
                budget = connection_manager.transient_budget(),
                current = connection_manager.transient_count(),
                "Transient connection budget exhausted; dropping inbound connection"
            );
            if let Some(callbacks) = state.awaiting_connection.remove(&peer_addr) {
                for mut cb in callbacks {
                    // Best effort - caller will handle the connection failure
                    #[allow(clippy::let_underscore_must_use)]
                    let _ = cb.send_result(Err(())).await;
                }
            }
            state.awaiting_connection_txs.remove(&peer_addr);
            return Ok(());
        }

        // IMPORTANT: Insert into self.connections BEFORE removing from awaiting_connection.
        // This prevents a race condition where another connect() call could slip through
        // during the window when neither structure has the entry, causing the transport
        // layer to tear down the connection we just established.
        // See: handle_connect_peer checks self.connections first, then awaiting_connection.
        // If the peer already has a connection (e.g., reconnected with new identity after
        // suspend/resume), replace it. Dropping the old sender causes its
        // peer_connection_listener to fire TransportClosed; the connection_id mechanism
        // ensures that stale event won't remove the new entry.
        if let Some(old) = self.connections.remove(&peer_addr) {
            if let Some(ref old_pub_key) = old.pub_key {
                self.addr_by_pub_key.remove(old_pub_key);
            }

            // Full cleanup: handle both transient and ring-promoted connections.
            // If transient, drop_transient releases the budget slot.
            // If ring-promoted, prune_connection cleans up topology, orphaned txs,
            // subscriptions, and notifies the handshake driver.
            let was_transient = self
                .bridge
                .op_manager
                .ring
                .connection_manager
                .drop_transient(peer_addr)
                .is_some();

            if !was_transient {
                // Old connection was ring-promoted — mirror TransportClosed cleanup
                let old_peer = if let Some(ref pub_key) = old.pub_key {
                    PeerKeyLocation::new(pub_key.clone(), peer_addr)
                } else {
                    PeerKeyLocation::new(
                        (*self.bridge.op_manager.ring.connection_manager.pub_key).clone(),
                        peer_addr,
                    )
                };
                let prune_result = self
                    .bridge
                    .op_manager
                    .ring
                    .prune_connection(PeerId::new(old_peer.pub_key().clone(), peer_addr))
                    .await;
                self.bridge
                    .handle_orphaned_transactions(
                        prune_result.orphaned_transactions,
                        &self.gateways,
                    )
                    .await;
                if prune_result.became_unready {
                    // Deferred: broadcast after connection_manager borrow ends
                    broadcast_ready = Some(false);
                }
                self.bridge
                    .op_manager
                    .on_ring_connection_lost(old_peer.pub_key());
                if let Err(error) = handshake_commands
                    .send(HandshakeCommand::DropConnection {
                        peer: old_peer.clone(),
                    })
                    .await
                {
                    tracing::warn!(
                        remote = %peer_addr,
                        ?error,
                        "Failed to notify handshake driver about replaced connection"
                    );
                }
            }

            tracing::info!(
                %peer_addr,
                old_connection_id = old.connection_id,
                was_transient,
                "Replacing stale connection entry (peer reconnected with new identity)"
            );
        }

        if is_transient {
            let cm = &self.bridge.op_manager.ring.connection_manager;
            let current = cm.transient_count();
            if current >= cm.transient_budget() {
                tracing::warn!(
                    remote = %peer_addr,
                    budget = cm.transient_budget(),
                    current,
                    "Transient connection budget exhausted; dropping inbound connection before insert"
                );
                return Ok(());
            }
        }
        let conn_id = next_connection_id();
        let (tx, rx) = mpsc::channel(10);
        tracing::debug!(
            self_peer = %self.bridge.op_manager.ring.connection_manager.pub_key,
            peer_id = ?peer_id,
            %peer_addr,
            connection_id = conn_id,
            conn_map_size = self.connections.len(),
            "[CONN_TRACK] INSERT: adding connection to HashMap"
        );
        self.connections.insert(
            peer_addr,
            ConnectionEntry {
                sender: tx,
                // For transient connections, we don't know the pub_key yet - it will be learned
                // when the peer sends its first message (e.g., ConnectRequest)
                pub_key: peer_id.as_ref().map(|p| p.pub_key().clone()),
                connection_id: conn_id,
                created_at: Instant::now(),
            },
        );
        // Only add to reverse lookup if we know the pub_key
        // For transient connections, this will be populated when identity is learned
        if let Some(ref peer) = peer_id {
            self.addr_by_pub_key
                .insert(peer.pub_key().clone(), peer_addr);
        }
        let Some(conn_events) = self.conn_event_tx.as_ref().cloned() else {
            anyhow::bail!("Connection event channel not initialized");
        };

        // Phase 4: Set orphan stream registry on connection for handling race conditions
        // between stream fragments and metadata messages (RequestStreaming/ResponseStreaming).
        let mut connection = connection;
        connection
            .set_orphan_stream_registry(self.bridge.op_manager.orphan_stream_registry().clone());

        // Use tokio::spawn directly instead of GlobalExecutor::spawn.
        // GlobalExecutor::spawn uses Handle::try_current().spawn() which doesn't
        // reliably poll tasks in certain test contexts (see issue #2709).
        tokio::spawn(async move {
            peer_connection_listener(rx, connection, peer_addr, conn_events, conn_id).await;
        });
        // Yield to allow the spawned peer_connection_listener task to start.
        // This is important because on some runtimes (especially in tests with boxed_local
        // futures), spawned tasks may not be scheduled immediately, causing messages
        // sent to the channel to pile up without being processed.
        tokio::task::yield_now().await;
        let newly_inserted = true;

        // Now safe to remove from awaiting_connection and notify callbacks.
        // self.connections already has the entry, so concurrent connect() calls
        // will see it and reuse the existing connection instead of tearing it down.
        let pending_txs = state
            .awaiting_connection_txs
            .remove(&peer_addr)
            .unwrap_or_default();
        if let Some(callbacks) = state.awaiting_connection.remove(&peer_addr) {
            // The callback expects the remote peer's address (not our own)
            let resolved_addr = peer_addr;
            tracing::debug!(
                remote = %peer_addr,
                callbacks = callbacks.len(),
                "handle_successful_connection: notifying waiting callbacks"
            );
            tracing::info!(
                remote = %peer_addr,
                callbacks = callbacks.len(),
                pending_txs = ?pending_txs,
                remaining_checks = ?remaining_checks,
                "handle_successful_connection: connection established"
            );
            for mut cb in callbacks {
                match timeout(
                    Duration::from_secs(60),
                    cb.send_result(Ok((resolved_addr, remaining_checks))),
                )
                .await
                {
                    Ok(Ok(())) => {}
                    Ok(Err(())) => {
                        tracing::debug!(
                            remote = %peer_addr,
                            "Callback dropped before receiving connection result"
                        );
                    }
                    Err(error) => {
                        tracing::error!(
                            remote = %peer_addr,
                            ?error,
                            "Failed to deliver connection result"
                        );
                    }
                }
            }
        } else {
            tracing::debug!(
                peer_id = ?peer_id,
                %peer_addr,
                pending_txs = ?pending_txs,
                "No callback for connection established"
            );
        }

        // Only promote to ring if we know the peer's identity.
        // For transient connections without known identity, we keep them as transport-only
        // until the peer identifies itself (e.g., via ConnectRequest).
        let promote_to_ring =
            peer_id.is_some() && (!is_transient || connection_manager.is_gateway());

        if newly_inserted {
            tracing::info!(peer_id = ?peer_id, %peer_addr, is_transient, "handle_successful_connection: inserted new connection entry");
            crate::node::network_status::record_peer_connected(peer_addr, None, None);
            if promote_to_ring {
                // Only prune reservation when promoting to ring - transient connections
                // don't go through should_accept() so they have no reservation to prune
                let pending_loc = connection_manager.prune_in_transit_connection(peer_addr);
                // Safe to unwrap: promote_to_ring is only true when peer_id.is_some()
                let peer = peer_id
                    .as_ref()
                    .expect("promote_to_ring requires known peer_id");
                let loc = pending_loc.unwrap_or_else(|| Location::from_address(&peer_addr));
                tracing::info!(
                    %peer_addr,
                    %loc,
                    pending_loc_known = pending_loc.is_some(),
                    "handle_successful_connection: evaluating promotion to ring"
                );
                // Don't re-run should_accept() here — the connection was already
                // accepted at the protocol level (the CONNECT terminus handler in connect.rs) and transport
                // establishment (NAT traversal) already succeeded. Re-evaluating
                // the probabilistic Kleinberg filter would discard hard-won
                // connections for no capacity reason (#3545).
                // Only enforce the hard max_connections cap below.
                let current = connection_manager.connection_count();
                if current >= connection_manager.max_connections {
                    tracing::warn!(
                        %peer_addr,
                        current_connections = current,
                        max_connections = connection_manager.max_connections,
                        %loc,
                        "handle_successful_connection: rejecting new connection to enforce cap"
                    );
                    // Drop the connection immediately instead of letting it linger
                    // as a zombie for up to 5 minutes confusing the dashboard.
                    crate::node::network_status::record_peer_disconnected(peer_addr);
                    self.drop_connection_by_addr(peer_addr, handshake_commands)
                        .await;
                    return Ok(());
                }
                tracing::info!(%peer_addr, %loc, "handle_successful_connection: promoting connection into ring");
                let just_became_ready = self
                    .bridge
                    .op_manager
                    .ring
                    .add_connection(loc, PeerId::new(peer.pub_key().clone(), peer_addr), true)
                    .await;
                let pkl = crate::ring::PeerKeyLocation::new(peer.pub_key().clone(), peer_addr);
                crate::node::network_status::record_peer_connected(
                    peer_addr,
                    Some(loc.as_f64()),
                    Some(pkl),
                );
                // Only count as NAT success for non-gateway peers (gateway connections are direct)
                if !crate::node::network_status::is_known_gateway(&peer_addr) {
                    crate::node::network_status::record_nat_attempt(true);
                }

                // tell the new peer about our subscriptions and cache
                for (target, msg) in self
                    .bridge
                    .op_manager
                    .on_ring_connection_established(peer_addr, peer.pub_key())
                {
                    if let Err(e) = self.bridge.send(target, msg).await {
                        tracing::warn!(
                            %peer_addr,
                            error = %e,
                            "Failed to send interest/cache data to new peer"
                        );
                    }
                }

                // Broadcast readiness if we just crossed the threshold (deferred)
                if just_became_ready {
                    broadcast_ready = Some(true);
                }

                // Note: In the simplified 2026-01 architecture, subscriptions are lease-based
                // and renewed periodically by the subscription renewal background task.
                // We no longer attempt to establish subscriptions opportunistically on peer connect.

                if is_transient {
                    connection_manager.drop_transient(peer_addr);
                }
            } else {
                // Not promoting to ring - either unknown identity or transient on non-gateway.
                //
                // IMPORTANT: Before registering as transient, check if the connection was already
                // promoted to ring by another code path (e.g., handle_connect_peer running during
                // our yield_now().await or callback sends). This prevents a race condition where:
                // 1. We start with peer_id=None, so promote_to_ring=false
                // 2. We yield at yield_now().await or cb.send_result().await
                // 3. Message arrives, handle_connect_peer promotes connection to ring
                // 4. We resume and would register as transient (re-inserting a dropped entry)
                // 5. TTL timer fires and removes the connection from ring!
                //
                // By checking has_connection_or_pending, we skip transient registration if the
                // connection was already promoted to ring during our await points.
                if connection_manager.has_connection_or_pending(peer_addr) {
                    tracing::debug!(
                        %peer_addr,
                        "Skipping transient registration - connection already in ring topology"
                    );
                } else {
                    // These connections don't go through should_accept() so there's no pending
                    // reservation. Compute location from address directly.
                    let loc = Location::from_address(&peer_addr);
                    connection_manager.try_register_transient(peer_addr, Some(loc));
                    tracing::info!(
                        peer_id = ?peer_id,
                        %peer_addr,
                        "Registered transient connection (not added to ring topology)"
                    );
                    let ttl = connection_manager.transient_ttl();
                    let cm = connection_manager.clone();
                    GlobalExecutor::spawn(async move {
                        sleep(ttl).await;
                        if cm.drop_transient(peer_addr).is_some() {
                            // Only remove the transient tracking entry. Do NOT
                            // dispatch DropConnection — that tears down the
                            // underlying transport, killing any in-progress
                            // CONNECT handshake. If the CONNECT succeeds later,
                            // handle_connect_peer() will detect the missing
                            // transient entry via is_in_ring() and still promote
                            // the peer to ring topology (see #3113 fix). If it
                            // never succeeds, the transport idle timeout handles
                            // cleanup.
                            tracing::info!(
                                %peer_addr,
                                "Transient connection expired; removed tracking entry \
                                 (transport preserved, handle_connect_peer will promote via fallback)"
                            );
                        }
                    });
                }
            }
        } else if is_transient {
            // We reserved budget earlier, but didn't take ownership of the connection.
            connection_manager.drop_transient(peer_addr);
        }
        // Deferred broadcast: connection_manager borrow is now dropped
        if let Some(ready) = broadcast_ready {
            self.handle_broadcast_ready_state(ready).await;
        }
        Ok(())
    }

    async fn handle_transport_event(
        &mut self,
        event: Option<ConnEvent>,
        state: &mut EventListenerState,
        handshake_commands: &HandshakeCommandSender,
    ) -> anyhow::Result<EventResult> {
        match event {
            Some(ConnEvent::InboundMessage(mut inbound)) => {
                let tx = *inbound.msg.id();

                if let Some(remote_addr) = inbound.remote_addr {
                    if let Some(sender_peer) = extract_sender_from_message(&inbound.msg) {
                        // Try to get known address, fall back to remote_addr if unknown
                        let sender_addr = KnownPeerKeyLocation::try_from(&sender_peer)
                            .ok()
                            .map(|k| k.socket_addr());
                        if sender_addr.map(|a| a == remote_addr).unwrap_or(false)
                            || sender_addr.map(|a| a.ip().is_unspecified()).unwrap_or(true)
                        {
                            let resolved_addr = {
                                let raw_addr = sender_addr.unwrap_or(remote_addr);
                                if raw_addr.ip().is_unspecified() {
                                    if let Some(sender_mut) =
                                        extract_sender_from_message_mut(&mut inbound.msg)
                                    {
                                        if sender_mut
                                            .socket_addr()
                                            .map(|a| a.ip().is_unspecified())
                                            .unwrap_or(true)
                                        {
                                            sender_mut.set_addr(remote_addr);
                                        }
                                    }
                                    remote_addr
                                } else {
                                    raw_addr
                                }
                            };
                            let new_peer_id =
                                PeerId::new(sender_peer.pub_key().clone(), resolved_addr);
                            // Check if we have a connection but with a different pub_key
                            if let Some(entry) = self.connections.get(&remote_addr) {
                                // If we don't have the pub_key stored yet or it differs from the new one, update it
                                let should_update = match &entry.pub_key {
                                    None => true,
                                    Some(old_pub_key) => old_pub_key != new_peer_id.pub_key(),
                                };
                                if should_update {
                                    let old_pub_key = entry.pub_key.clone();
                                    let is_first_identity = old_pub_key.is_none();
                                    tracing::info!(
                                        remote = %remote_addr,
                                        old_pub_key = ?old_pub_key,
                                        new_pub_key = %new_peer_id.pub_key(),
                                        is_first_identity,
                                        "Updating peer identity after inbound message"
                                    );
                                    // Remove old reverse lookup if it exists
                                    if let Some(old_key) = old_pub_key {
                                        self.addr_by_pub_key.remove(&old_key);
                                        // Update ring with old PeerId -> new PeerId
                                        let old_peer_id = PeerId::new(old_key, remote_addr);
                                        self.bridge.op_manager.ring.update_connection_identity(
                                            &old_peer_id,
                                            new_peer_id.clone(),
                                        );
                                    }
                                    // Update the entry's pub_key
                                    if let Some(entry) = self.connections.get_mut(&remote_addr) {
                                        entry.pub_key = Some(new_peer_id.pub_key().clone());
                                    }
                                    // Add new reverse lookup
                                    self.addr_by_pub_key
                                        .insert(new_peer_id.pub_key().clone(), remote_addr);
                                    // Note: We do NOT automatically promote to ring here.
                                    // Transient connections are promoted only when the Connect
                                    // operation explicitly accepts via NodeEvent::ConnectPeer,
                                    // which is handled by handle_connect_peer().
                                }
                            }
                        }
                    }
                }

                tracing::debug!(
                    peer_addr = ?inbound.remote_addr,
                    %tx,
                    tx_type = ?tx.transaction_type(),
                    "Queueing inbound NetMessage from peer connection"
                );
                Ok(EventResult::Event(
                    ConnEvent::InboundMessage(inbound).into(),
                ))
            }
            Some(ConnEvent::TransportClosed {
                remote_addr,
                error,
                connection_id,
            }) => {
                tracing::debug!(
                    remote = %remote_addr,
                    ?error,
                    connection_id,
                    "peer_connection_listener reported transport closure"
                );
                // Ignore stale TransportClosed from replaced connections: the old
                // listener's channel was dropped, but we must not remove the new entry.
                if let Some(current) = self.connections.get(&remote_addr) {
                    if current.connection_id != connection_id {
                        tracing::info!(
                            remote = %remote_addr,
                            stale_id = connection_id,
                            current_id = current.connection_id,
                            "Ignoring stale TransportClosed from replaced connection"
                        );
                        return Ok(EventResult::Continue);
                    }
                }
                // Look up the connection directly by address
                if let Some(entry) = self.connections.remove(&remote_addr) {
                    // Construct PeerKeyLocation for prune_connection and DropConnection
                    let peer = if let Some(ref pub_key) = entry.pub_key {
                        PeerKeyLocation::new(pub_key.clone(), remote_addr)
                    } else {
                        PeerKeyLocation::new(
                            (*self.bridge.op_manager.ring.connection_manager.pub_key).clone(),
                            remote_addr,
                        )
                    };
                    // Remove from reverse lookup
                    if let Some(pub_key) = entry.pub_key {
                        self.addr_by_pub_key.remove(&pub_key);
                    }
                    tracing::debug!(self_peer = %self.bridge.op_manager.ring.connection_manager.pub_key, %peer, socket_addr = %remote_addr, conn_map_size = self.connections.len(), "[CONN_TRACK] REMOVE: TransportClosed - removing from connections HashMap");
                    let prune_result = self
                        .bridge
                        .op_manager
                        .ring
                        .prune_connection(PeerId::new(peer.pub_key().clone(), remote_addr))
                        .await;

                    // Handle orphaned transactions immediately (retry via alternate routes)
                    self.bridge
                        .handle_orphaned_transactions(
                            prune_result.orphaned_transactions,
                            &self.gateways,
                        )
                        .await;

                    if prune_result.became_unready {
                        self.handle_broadcast_ready_state(false).await;
                    }

                    // forget this peer's subscriptions and cached contract state
                    self.bridge
                        .op_manager
                        .on_ring_connection_lost(peer.pub_key());

                    // Backoff prevents reconnect-timeout-reconnect cycles to dead peers (#3252).
                    if !remote_addr.ip().is_unspecified() {
                        state.peer_backoff.record_failure(remote_addr);
                        tracing::debug!(remote = %remote_addr, "Recorded peer backoff after transport closure");
                    }

                    if let Err(error) = handshake_commands
                        .send(HandshakeCommand::DropConnection { peer: peer.clone() })
                        .await
                    {
                        tracing::warn!(
                            remote = %remote_addr,
                            ?error,
                            "Failed to notify handshake driver about dropped connection"
                        );
                    }
                }
                Ok(EventResult::Continue)
            }
            Some(other) => {
                tracing::warn!(?other, "Unexpected event from peer connection listener");
                Ok(EventResult::Continue)
            }
            None => {
                tracing::error!("All peer connection event channels closed");
                Ok(EventResult::Continue)
            }
        }
    }

    fn handle_notification_msg(&self, msg: Option<Either<NetMessage, NodeEvent>>) -> EventResult {
        match msg {
            Some(Left(msg)) => {
                // With hop-by-hop routing, messages no longer embed target.
                // For initial requests (GET, PUT, Subscribe, Connect), extract next hop from operation state.
                // For other messages, process locally (they'll be routed through network_bridge.send()).
                let tx = *msg.id();

                // Try to get next hop from operation state for initial outbound requests
                // Uses peek methods to avoid pop/push overhead
                if let Some(next_hop_addr) = self.bridge.op_manager.peek_next_hop_addr(&tx) {
                    tracing::debug!(
                        tx = %msg.id(),
                        msg_type = %msg,
                        next_hop = %next_hop_addr,
                        "handle_notification_msg: Found next hop in operation state, routing as OutboundMessage"
                    );
                    // Fire-and-forget ops (Update, Unsubscribe) are completed after routing.
                    // Other ops expect responses and stay in the state map.
                    let is_fire_and_forget = tx.transaction_type() == TransactionType::Update
                        || matches!(
                            &msg,
                            NetMessage::V1(NetMessageV1::Subscribe(
                                crate::operations::subscribe::SubscribeMsg::Unsubscribe { .. }
                            ))
                        );
                    if is_fire_and_forget {
                        self.bridge.op_manager.completed(tx);
                    }
                    return EventResult::Event(
                        ConnEvent::OutboundMessageWithTarget {
                            target_addr: next_hop_addr,
                            msg,
                        }
                        .into(),
                    );
                }

                // Message has no next hop or couldn't get from op state - process locally
                tracing::debug!(
                    tx = %msg.id(),
                    msg_type = %msg,
                    "handle_notification_msg: No next hop found, processing locally"
                );
                EventResult::Event(ConnEvent::InboundMessage(msg.into()).into())
            }
            Some(Right(action)) => {
                tracing::debug!(
                    event = %action,
                    "handle_notification_msg: Received NodeEvent notification"
                );
                EventResult::Event(ConnEvent::NodeAction(action).into())
            }
            None => EventResult::Event(
                ConnEvent::ClosedChannel(ChannelCloseReason::Notification).into(),
            ),
        }
    }

    fn handle_op_execution(
        &self,
        msg: Option<(Sender<NetMessage>, NetMessage)>,
        state: &mut EventListenerState,
    ) -> EventResult {
        match msg {
            Some((callback, msg)) => {
                state.pending_op_results.insert(*msg.id(), callback);
                crate::config::GlobalTestMetrics::record_pending_op_insert();
                crate::config::GlobalTestMetrics::record_pending_op_size(
                    state.pending_op_results.len() as u64,
                );
                EventResult::Event(ConnEvent::InboundMessage(msg.into()).into())
            }
            None => {
                EventResult::Event(ConnEvent::ClosedChannel(ChannelCloseReason::OpExecution).into())
            }
        }
    }

    fn handle_bridge_msg(&self, msg: Option<P2pBridgeEvent>) -> EventResult {
        match msg {
            Some(P2pBridgeEvent::Message(target, msg)) => {
                // Use OutboundMessageWithTarget to preserve the target address from
                // P2pBridge::send(). This is critical for NAT scenarios where
                // the address in the message differs from the actual transport address.
                // The target.socket_addr() contains the address that was used to look up
                // the peer in P2pBridge::send(), which is the correct transport address.
                if let Some(target_addr) = target.socket_addr() {
                    EventResult::Event(
                        ConnEvent::OutboundMessageWithTarget {
                            target_addr,
                            msg: *msg,
                        }
                        .into(),
                    )
                } else {
                    // Fall back to OutboundMessage if no explicit address
                    // (shouldn't happen in normal operation)
                    tracing::warn!(
                        tx = %msg.id(),
                        target_pub_key = %target.pub_key(),
                        "handle_bridge_msg: target has no socket address, falling back to msg.target()"
                    );
                    EventResult::Event(ConnEvent::OutboundMessage(*msg).into())
                }
            }
            Some(P2pBridgeEvent::NodeAction(action)) => {
                EventResult::Event(ConnEvent::NodeAction(action).into())
            }
            Some(P2pBridgeEvent::StreamSend {
                target_addr,
                stream_id,
                data,
                metadata,
                completion_tx,
            }) => EventResult::Event(
                ConnEvent::StreamSend {
                    target_addr,
                    stream_id,
                    data,
                    metadata,
                    completion_tx,
                }
                .into(),
            ),
            Some(P2pBridgeEvent::PipeStream {
                target_addr,
                outbound_stream_id,
                inbound_handle,
                metadata,
            }) => EventResult::Event(
                ConnEvent::PipeStream {
                    target_addr,
                    outbound_stream_id,
                    inbound_handle,
                    metadata,
                }
                .into(),
            ),
            None => EventResult::Event(ConnEvent::ClosedChannel(ChannelCloseReason::Bridge).into()),
        }
    }

    fn handle_node_controller_msg(&self, msg: Option<NodeEvent>) -> EventResult {
        match msg {
            Some(msg) => EventResult::Event(ConnEvent::NodeAction(msg).into()),
            None => {
                EventResult::Event(ConnEvent::ClosedChannel(ChannelCloseReason::Controller).into())
            }
        }
    }

    // Removed handle_handshake_msg as it's integrated into wait_for_event

    fn handle_client_transaction_subscription(
        &self,
        event_id: Result<(ClientId, WaitingTransaction), anyhow::Error>,
        state: &mut EventListenerState,
    ) -> EventResult {
        let Ok((client_id, transaction)) = event_id.inspect_err(|e| {
            tracing::error!("Error while receiving client transaction result: {:?}", e);
        }) else {
            return EventResult::Continue;
        };
        match transaction {
            WaitingTransaction::Transaction(tx) => {
                tracing::debug!(%tx, %client_id, "Subscribing client to transaction results");
                let entry = state.tx_to_client.entry(tx).or_default();
                let inserted = entry.insert(client_id);
                tracing::debug!(
                    "tx_to_client: tx={} client={} inserted={} total_waiting_clients={}",
                    tx,
                    client_id,
                    inserted,
                    entry.len()
                );
            }
            WaitingTransaction::Subscription { contract_key } => {
                tracing::debug!(%client_id, %contract_key, "Client waiting for subscription");
                if let Some(clients) =
                    state
                        .client_waiting_transaction
                        .iter_mut()
                        .find_map(|(tx, clients)| {
                            if let WaitingTransaction::Subscription { contract_key: key } = tx {
                                return (key == &contract_key).then_some(clients);
                            }
                            None
                        })
                {
                    clients.insert(client_id);
                } else {
                    state.client_waiting_transaction.push((
                        WaitingTransaction::Subscription { contract_key },
                        HashSet::from_iter([client_id]),
                    ));
                }
            }
        }
        EventResult::Continue
    }

    fn handle_executor_transaction(
        &self,
        id: Result<Transaction, anyhow::Error>,
        state: &mut EventListenerState,
    ) -> EventResult {
        let Ok(id) = id.map_err(|err| {
            tracing::error!("Error while receiving transaction from executor: {:?}", err);
        }) else {
            return EventResult::Continue;
        };
        state.pending_from_executor.insert(id);
        EventResult::Continue
    }

    /// Broadcast a message to a list of peers off the event loop.
    ///
    /// In production, the sends are spawned via `tokio::spawn` to prevent
    /// self-deadlock: the bridge channel is drained by the event loop, so
    /// inline fan-out of >capacity sends would block forever.
    /// In simulation tests, sends run inline for determinism.
    async fn broadcast_to_peers(
        bridge: &P2pBridge,
        peers: Vec<SocketAddr>,
        msg: crate::message::NetMessage,
    ) {
        let bridge = bridge.clone();

        let send_all = async move {
            for peer_addr in peers {
                if let Err(e) = bridge.send(peer_addr, msg.clone()).await {
                    tracing::warn!(
                        peer_addr = %peer_addr,
                        error = %e,
                        "Failed to send broadcast message to peer"
                    );
                }
            }
        };

        #[cfg(not(feature = "simulation_tests"))]
        tokio::spawn(send_all);

        #[cfg(feature = "simulation_tests")]
        send_all.await;
    }

    /// Broadcast a hosting update message to all connected peers.
    async fn handle_hosting_broadcast(&mut self, message: crate::message::NeighborHostingMessage) {
        tracing::debug!(
            message = ?message,
            peer_count = self.connections.len(),
            phase = "broadcast",
            "Broadcasting hosting update to connected peers"
        );

        let msg = crate::message::NetMessage::V1(crate::message::NetMessageV1::NeighborHosting {
            message: message.clone(),
        });
        let peers: Vec<SocketAddr> = self.connections.keys().copied().collect();
        Self::broadcast_to_peers(&self.bridge, peers, msg).await;
    }

    /// Broadcast ChangeInterests message to all connected peers.
    async fn handle_broadcast_change_interests(&mut self, added: Vec<u32>, removed: Vec<u32>) {
        tracing::debug!(
            added_count = added.len(),
            removed_count = removed.len(),
            peer_count = self.connections.len(),
            "Broadcasting ChangeInterests to connected peers"
        );

        let msg = crate::message::NetMessage::V1(crate::message::NetMessageV1::InterestSync {
            message: crate::message::InterestMessage::ChangeInterests { added, removed },
        });
        let peers: Vec<SocketAddr> = self.connections.keys().copied().collect();
        Self::broadcast_to_peers(&self.bridge, peers, msg).await;
    }

    /// Send an interest message to a specific peer.
    async fn handle_send_interest_message(
        &mut self,
        target: SocketAddr,
        message: crate::message::InterestMessage,
    ) {
        tracing::debug!(
            target = %target,
            "Sending interest message to peer"
        );

        let msg =
            crate::message::NetMessage::V1(crate::message::NetMessageV1::InterestSync { message });

        if let Err(e) = self.bridge.send(target, msg).await {
            tracing::warn!(
                peer_addr = %target,
                error = %e,
                "Failed to send interest message to peer"
            );
        }
    }

    /// Broadcast ReadyState to all connected ring peers.
    async fn handle_broadcast_ready_state(&mut self, ready: bool) {
        let msg =
            crate::message::NetMessage::V1(crate::message::NetMessageV1::ReadyState { ready });

        tracing::info!(
            ready,
            peer_count = self.connections.len(),
            "Broadcasting ReadyState to connected peers"
        );

        let peers: Vec<SocketAddr> = self.connections.keys().copied().collect();
        Self::broadcast_to_peers(&self.bridge, peers, msg).await;
    }

    /// Maximum retry attempts when a broadcast finds no targets.
    const MAX_BROADCAST_RETRIES: u8 = 3;

    /// Base delay between broadcast retries (scaled by attempt number for linear backoff).
    const BROADCAST_RETRY_BASE_DELAY: Duration = Duration::from_secs(1);

    /// Maximum entries in the no-target streak tracker. Prevents unbounded growth
    /// from network-influenced contract keys.
    const MAX_BROADCAST_STREAK_ENTRIES: usize = 256;

    /// Notify interested network peers about a state change.
    ///
    /// Echo-back is prevented by summary comparison: we skip peers whose cached
    /// summary matches ours (they already have our state).
    ///
    /// When no targets are found (race between contract updates and subscription
    /// establishment), the broadcast is retried with linear backoff up to
    /// [`Self::MAX_BROADCAST_RETRIES`] times.
    async fn handle_broadcast_state_change(
        &mut self,
        op_manager: &Arc<OpManager>,
        key: freenet_stdlib::prelude::ContractKey,
        new_state: freenet_stdlib::prelude::WrappedState,
    ) {
        tracing::debug!(
            contract = %key,
            state_size = new_state.size(),
            "BroadcastStateChange event received"
        );
        let self_addr = op_manager.ring.connection_manager.get_own_addr();
        let Some(self_addr) = self_addr else {
            tracing::warn!(
                contract = %key,
                "Cannot broadcast state change - no own address"
            );
            return;
        };

        let target_result = op_manager.get_broadcast_targets_update(&key, &self_addr);
        tracing::debug!(
            contract = %key,
            target_count = target_result.targets.len(),
            self_addr = %self_addr,
            "BroadcastStateChange: found targets"
        );

        if target_result.targets.is_empty() {
            let retry_count = self.broadcast_retries.entry(key).or_insert(0);
            if *retry_count < Self::MAX_BROADCAST_RETRIES {
                *retry_count += 1;
                let attempt = *retry_count;
                tracing::info!(
                    contract = %key,
                    self_addr = %self_addr,
                    attempt,
                    max_retries = Self::MAX_BROADCAST_RETRIES,
                    "BROADCAST_NO_TARGETS: scheduling retry - subscriptions may still be in-flight"
                );
                // Schedule a delayed re-emission of BroadcastStateChange
                let op_mgr = op_manager.clone();
                let delay = Self::BROADCAST_RETRY_BASE_DELAY * u32::from(attempt);
                tokio::spawn(async move {
                    tokio::time::sleep(delay).await;
                    if let Err(e) = op_mgr
                        .notify_node_event(crate::message::NodeEvent::BroadcastStateChange {
                            key,
                            new_state,
                        })
                        .await
                    {
                        tracing::warn!(
                            contract = %key,
                            error = %e,
                            "Failed to re-emit BroadcastStateChange for retry"
                        );
                    }
                });
            } else {
                // Retries exhausted. Track consecutive no-target cycles to
                // suppress repetitive WARN logging after the first occurrence.
                self.broadcast_retries.remove(&key);

                // Evict oldest entry if at capacity to prevent unbounded growth.
                if !self.broadcast_no_target_streak.contains_key(&key)
                    && self.broadcast_no_target_streak.len() >= Self::MAX_BROADCAST_STREAK_ENTRIES
                {
                    if let Some(evict_key) = self.broadcast_no_target_streak.keys().next().cloned()
                    {
                        self.broadcast_no_target_streak.remove(&evict_key);
                    }
                }
                let streak = self.broadcast_no_target_streak.entry(key).or_insert(0);
                *streak = streak.saturating_add(1);
                let current_streak = *streak;

                if current_streak <= 1 {
                    tracing::warn!(
                        contract = %key,
                        self_addr = %self_addr,
                        "BROADCAST_NO_TARGETS: no targets found after {} retries, giving up",
                        Self::MAX_BROADCAST_RETRIES
                    );
                } else {
                    tracing::debug!(
                        contract = %key,
                        self_addr = %self_addr,
                        consecutive_misses = current_streak,
                        "BROADCAST_NO_TARGETS: still no targets (suppressing repeated warns)"
                    );
                }

                // Emit delivery summary for diagnostics (only on first miss per streak)
                if current_streak <= 1 {
                    let update_tx =
                        crate::message::Transaction::new::<crate::operations::update::UpdateMsg>();
                    if let Some(log) = NetEventLog::broadcast_delivery_summary(
                        &update_tx,
                        &op_manager.ring,
                        key,
                        &target_result,
                        0,
                        0,
                        0,
                    ) {
                        self.bridge
                            .log_register
                            .register_events(Either::Left(log))
                            .await;
                    }
                }
            }
            return;
        }

        // Targets found - clear any pending retry and streak state for this contract
        self.broadcast_retries.remove(&key);
        self.broadcast_no_target_streak.remove(&key);

        // In production, enqueue each target into the broadcast queue.
        // The queue worker handles delta computation and streaming with bounded
        // concurrency (default: 2 concurrent streams) to prevent uplink saturation.
        //
        // In simulation tests, call inline to preserve deterministic message
        // ordering — tokio::spawn in start_paused(true) changes broadcast
        // delivery order and breaks convergence.
        #[cfg(not(feature = "simulation_tests"))]
        {
            for target in &target_result.targets {
                self.broadcast_queue
                    .enqueue(key, target.clone(), new_state.clone())
                    .await;
            }
            // Emit broadcast emitted telemetry (issue #3622)
            //
            // Production path limitations:
            // - Transaction ID is synthetic (not from the original update operation).
            //   BroadcastQueue creates its own TX per target, so this ID cannot be
            //   correlated with downstream BroadcastReceived/BroadcastApplied events.
            // - `broadcasted_to` = number of targets enqueued, not actual delivery
            //   count. Actual sends are async via BroadcastQueue.
            // - `broadcast_to` is empty to avoid cloning up to 512 PeerKeyLocations
            //   purely for telemetry. The peer list can be reconstructed from
            //   BroadcastReceived events on the receiving end.
            let update_tx =
                crate::message::Transaction::new::<crate::operations::update::UpdateMsg>();
            let enqueued_count = target_result.targets.len();
            if let Some(log) = NetEventLog::update_broadcast_emitted(
                &update_tx,
                &op_manager.ring,
                key,
                Vec::new(),
                enqueued_count,
                new_state,
            ) {
                self.bridge
                    .log_register
                    .register_events(Either::Left(log))
                    .await;
            }
        }
        #[cfg(feature = "simulation_tests")]
        {
            Self::broadcast_state_to_peers(
                self.bridge.clone(),
                op_manager,
                key,
                new_state,
                target_result,
            )
            .await;
        }
    }

    /// Send state to a specific peer that reported a stale summary.
    ///
    /// Unlike `handle_broadcast_state_change` which fans out to ALL subscribers,
    /// this targets only the peer that needs catching up. Used by the interest
    /// sync summary-mismatch handler to avoid O(peers^2) broadcast storms.
    async fn handle_sync_state_to_peer(
        &mut self,
        op_manager: &Arc<OpManager>,
        key: freenet_stdlib::prelude::ContractKey,
        new_state: freenet_stdlib::prelude::WrappedState,
        target_addr: std::net::SocketAddr,
    ) {
        let target = op_manager
            .ring
            .connection_manager
            .get_peer_by_addr(target_addr);
        let Some(target) = target else {
            tracing::debug!(
                contract = %key,
                peer = %target_addr,
                "SyncStateToPeer: peer not found in connection manager, skipping"
            );
            return;
        };

        tracing::debug!(
            contract = %key,
            peer = %target_addr,
            "SyncStateToPeer: sending state to stale peer"
        );

        #[cfg(not(feature = "simulation_tests"))]
        {
            self.broadcast_queue.enqueue(key, target, new_state).await;
        }
        #[cfg(feature = "simulation_tests")]
        {
            super::broadcast_queue::broadcast_to_single_peer(
                &self.bridge,
                op_manager,
                key,
                new_state,
                target,
            )
            .await;
        }
    }

    /// Send state change broadcasts to interested peers.
    ///
    /// Used only in simulation tests for inline (deterministic) broadcast ordering.
    /// In production, the `BroadcastQueue` handles per-target sends with bounded
    /// concurrency.
    #[cfg(feature = "simulation_tests")]
    async fn broadcast_state_to_peers(
        bridge: P2pBridge,
        op_manager: &Arc<OpManager>,
        key: freenet_stdlib::prelude::ContractKey,
        new_state: freenet_stdlib::prelude::WrappedState,
        target_result: crate::operations::update::BroadcastTargetResult,
    ) {
        // Get our summary once for all targets
        let our_summary = op_manager
            .interest_manager
            .get_contract_summary(op_manager, &key)
            .await;

        let update_tx = crate::message::Transaction::new::<crate::operations::update::UpdateMsg>();

        tracing::debug!(
            tx = %update_tx,
            contract = %key,
            target_count = target_result.targets.len(),
            "Broadcasting state change to network peers"
        );

        let mut skipped_summary_match: usize = 0;
        let mut send_success: usize = 0;
        let mut send_failed: usize = 0;

        for target in &target_result.targets {
            let Some(peer_addr) = target.socket_addr() else {
                continue;
            };

            let peer_key = PeerKey::from(target.pub_key().clone());

            // Get peer's cached summary
            let their_summary = op_manager
                .interest_manager
                .get_peer_summary(&key, &peer_key);

            // Skip if summaries are equal (no change to send)
            if let (Some(ours), Some(theirs)) = (&our_summary, &their_summary) {
                if ours.as_ref() == theirs.as_ref() {
                    tracing::trace!(
                        contract = %key,
                        peer = %peer_addr,
                        "Skipping broadcast - peer already has our state"
                    );
                    skipped_summary_match += 1;
                    continue;
                }
            }

            // Compute delta if we have their summary (uses memoization cache)
            // Track whether we successfully computed a delta vs sent full state
            let (payload, sent_delta) = match (&our_summary, &their_summary) {
                (Some(ours), Some(theirs)) => {
                    match op_manager
                        .interest_manager
                        .compute_delta(op_manager, &key, theirs, ours, new_state.size())
                        .await
                    {
                        Ok(Some(delta)) => (
                            crate::message::DeltaOrFullState::Delta(delta.as_ref().to_vec()),
                            true,
                        ),
                        Ok(None) => {
                            tracing::debug!(
                                contract = %key,
                                "Delta computation returned no change, sending full state"
                            );
                            (
                                crate::message::DeltaOrFullState::FullState(
                                    new_state.as_ref().to_vec(),
                                ),
                                false,
                            )
                        }
                        Err(err) => {
                            tracing::debug!(
                                contract = %key,
                                error = %err,
                                "Delta computation failed, falling back to full state"
                            );
                            (
                                crate::message::DeltaOrFullState::FullState(
                                    new_state.as_ref().to_vec(),
                                ),
                                false,
                            )
                        }
                    }
                }
                _ => (
                    crate::message::DeltaOrFullState::FullState(new_state.as_ref().to_vec()),
                    false,
                ),
            };

            // Save payload size before moving it into the message
            let payload_size = payload.size();

            // Check if we should use streaming for full state broadcasts
            let use_streaming = matches!(&payload, crate::message::DeltaOrFullState::FullState(_))
                && crate::operations::should_use_streaming(
                    op_manager.streaming_threshold,
                    payload_size,
                );

            let send_result = if use_streaming {
                let sender_summary_bytes = our_summary
                    .as_ref()
                    .map(|s| s.as_ref().to_vec())
                    .unwrap_or_default();
                let state_bytes = match payload {
                    crate::message::DeltaOrFullState::FullState(data) => data,
                    _ => unreachable!("checked above"),
                };
                let streaming_payload = crate::operations::update::BroadcastStreamingPayload {
                    state_bytes,
                    sender_summary_bytes,
                };
                let payload_bytes = match bincode::serialize(&streaming_payload) {
                    Ok(b) => b,
                    Err(e) => {
                        tracing::warn!(
                            tx = %update_tx,
                            error = %e,
                            "Failed to serialize BroadcastStreamingPayload, skipping"
                        );
                        continue;
                    }
                };
                let sid = StreamId::next_operations();
                tracing::debug!(
                    tx = %update_tx,
                    contract = %key,
                    peer = %peer_addr,
                    stream_id = %sid,
                    payload_size,
                    "Using streaming for BroadcastTo"
                );
                let msg = crate::operations::update::UpdateMsg::BroadcastToStreaming {
                    id: update_tx,
                    stream_id: sid,
                    key,
                    total_size: payload_bytes.len() as u64,
                };
                let net_msg: NetMessage = msg.into();
                // Serialize metadata for embedding in fragment #1 (fix #2757)
                let metadata = match bincode::serialize(&net_msg) {
                    Ok(bytes) => Some(bytes::Bytes::from(bytes)),
                    Err(e) => {
                        tracing::warn!(
                            ?peer_addr,
                            error = %e,
                            "Failed to serialize BroadcastTo metadata for embedding"
                        );
                        None
                    }
                };
                let send_res = bridge.send(peer_addr, net_msg).await;
                if send_res.is_ok() {
                    // Send the stream data after the metadata message
                    if let Err(err) = bridge
                        .send_stream(peer_addr, sid, bytes::Bytes::from(payload_bytes), metadata)
                        .await
                    {
                        tracing::warn!(
                            tx = %update_tx,
                            peer = %peer_addr,
                            error = %err,
                            "Failed to send broadcast stream data"
                        );
                    }
                }
                send_res
            } else {
                let msg = crate::operations::update::UpdateMsg::BroadcastTo {
                    id: update_tx,
                    key,
                    payload,
                    // Include our summary so peer doesn't echo back
                    sender_summary_bytes: our_summary
                        .as_ref()
                        .map(|s| s.as_ref().to_vec())
                        .unwrap_or_default(),
                };
                bridge.send(peer_addr, msg.into()).await
            };

            if let Err(err) = send_result {
                send_failed += 1;
                tracing::warn!(
                    tx = %update_tx,
                    peer = %peer_addr,
                    error = %err,
                    "Failed to send state change broadcast"
                );
            } else {
                send_success += 1;
                // Track delta vs full state sends for testing (PR #2763)
                if sent_delta {
                    op_manager
                        .interest_manager
                        .record_delta_send(new_state.size(), payload_size);
                    crate::config::GlobalTestMetrics::record_delta_send();
                } else {
                    op_manager.interest_manager.record_full_state_send();
                    crate::config::GlobalTestMetrics::record_full_state_send();
                }

                // Issue #3046: Refresh the peer's interest TTL on every successful
                // broadcast send. Without this, interest entries for peers who only
                // receive full-state broadcasts (no delta → no update_peer_summary
                // call) expire after INTEREST_TTL, even though broadcasts
                // are being successfully delivered. This caused ~49% of River room
                // subscribers to miss updates.
                op_manager
                    .interest_manager
                    .refresh_peer_interest(&key, &peer_key);
            }

            // PR #2763 FIX: Only update cached summary when we sent a delta.
            // This is separate from the TTL refresh above — update_peer_summary
            // sets the cached summary (used for delta computation), while
            // refresh_peer_interest only extends the TTL.
            if sent_delta {
                if let Some(summary) = &our_summary {
                    op_manager.interest_manager.update_peer_summary(
                        &key,
                        &peer_key,
                        Some(summary.clone()),
                    );
                }
            }
        }

        // Emit broadcast emitted telemetry (issue #3622)
        if let Some(log) = NetEventLog::update_broadcast_emitted(
            &update_tx,
            &op_manager.ring,
            key,
            target_result.targets.clone(),
            send_success,
            new_state,
        ) {
            bridge.log_register.register_events(Either::Left(log)).await;
        }

        // Emit broadcast delivery summary telemetry (issue #3046)
        if let Some(log) = NetEventLog::broadcast_delivery_summary(
            &update_tx,
            &op_manager.ring,
            key,
            &target_result,
            skipped_summary_match,
            send_success,
            send_failed,
        ) {
            bridge.log_register.register_events(Either::Left(log)).await;
        }
    }
}

trait ConnectResultSender: Send {
    fn send_result(
        &mut self,
        result: Result<(SocketAddr, Option<usize>), ()>,
    ) -> Pin<Box<dyn Future<Output = Result<(), ()>> + Send + '_>>;
}

impl ConnectResultSender for mpsc::Sender<Result<(SocketAddr, Option<usize>), ()>> {
    fn send_result(
        &mut self,
        result: Result<(SocketAddr, Option<usize>), ()>,
    ) -> Pin<Box<dyn Future<Output = Result<(), ()>> + Send + '_>> {
        async move { self.send(result).await.map_err(|_| ()) }.boxed()
    }
}

struct EventListenerState {
    expected_inbound: ExpectedInboundTracker,
    pending_from_executor: HashSet<Transaction>,
    /// Maps transactions to the set of clients waiting for their results.
    /// Cleaned up via `TransactionTimedOut` and `TransactionCompleted` handlers.
    tx_to_client: HashMap<Transaction, HashSet<ClientId>>,
    client_waiting_transaction: Vec<(WaitingTransaction, HashSet<ClientId>)>,
    awaiting_connection: HashMap<SocketAddr, Vec<Box<dyn ConnectResultSender>>>,
    awaiting_connection_txs: HashMap<SocketAddr, Vec<Transaction>>,
    pending_op_results: HashMap<Transaction, Sender<NetMessage>>,
    /// Last time pending_op_results was scanned for closed senders.
    last_pending_op_cleanup: Instant,
    /// Per-peer backoff tracking for failed connection attempts.
    /// Prevents rapid repeated connection attempts to the same peer.
    peer_backoff: PeerConnectionBackoff,
    /// Last time peer_backoff was cleaned up.
    last_backoff_cleanup: Instant,
}

impl EventListenerState {
    fn new(expected_inbound: ExpectedInboundTracker) -> Self {
        Self {
            expected_inbound,
            pending_from_executor: HashSet::new(),
            tx_to_client: HashMap::new(),
            client_waiting_transaction: Vec::new(),
            awaiting_connection: HashMap::new(),
            pending_op_results: HashMap::new(),
            last_pending_op_cleanup: Instant::now(),
            awaiting_connection_txs: HashMap::new(),
            peer_backoff: PeerConnectionBackoff::new(),
            last_backoff_cleanup: Instant::now(),
        }
    }
}

enum EventResult {
    Continue,
    Event(Box<ConnEvent>),
}

#[derive(Debug)]
pub(super) enum ConnEvent {
    InboundMessage(IncomingMessage),
    OutboundMessage(NetMessage),
    /// Outbound message with explicit target address from P2pBridge::send().
    /// Used when the target address differs from what's in the message (NAT scenarios).
    /// The target_addr is the actual transport address to send to.
    OutboundMessageWithTarget {
        target_addr: SocketAddr,
        msg: NetMessage,
    },
    NodeAction(NodeEvent),
    ClosedChannel(ChannelCloseReason),
    TransportClosed {
        remote_addr: SocketAddr,
        error: TransportError,
        /// ID of the connection entry that spawned the listener reporting this closure.
        /// Used to ignore stale events from replaced connections.
        connection_id: u64,
    },
    /// Send raw stream data to a peer via the transport's stream mechanism.
    /// Used by operations-level streaming to send large payloads as stream fragments.
    StreamSend {
        target_addr: SocketAddr,
        stream_id: StreamId,
        data: bytes::Bytes,
        metadata: Option<bytes::Bytes>,
        /// Optional completion signal for broadcast queue concurrency control.
        completion_tx: Option<tokio::sync::oneshot::Sender<()>>,
    },
    /// Pipe an inbound stream to a peer, forwarding fragments as they arrive.
    /// Used for low-latency forwarding at intermediate nodes.
    PipeStream {
        target_addr: SocketAddr,
        outbound_stream_id: StreamId,
        inbound_handle: crate::transport::peer_connection::streaming::StreamHandle,
        metadata: Option<bytes::Bytes>,
    },
}

#[derive(Debug)]
pub(super) struct IncomingMessage {
    pub msg: NetMessage,
    pub remote_addr: Option<SocketAddr>,
}

impl IncomingMessage {
    fn with_remote(msg: NetMessage, remote_addr: SocketAddr) -> Self {
        Self {
            msg,
            remote_addr: Some(remote_addr),
        }
    }
}

impl From<NetMessage> for IncomingMessage {
    fn from(msg: NetMessage) -> Self {
        Self {
            msg,
            remote_addr: None,
        }
    }
}

#[derive(Debug)]
pub(super) enum ChannelCloseReason {
    /// Internal bridge channel closed - critical, must shutdown gracefully
    Bridge,
    /// Node controller channel closed - critical, must shutdown gracefully
    Controller,
    /// Notification channel closed - critical, must shutdown gracefully
    Notification,
    /// Op execution channel closed - critical, must shutdown gracefully
    OpExecution,
}

#[allow(dead_code)]
enum ProtocolStatus {
    Unconfirmed,
    Confirmed,
    Reported,
    Failed,
}

async fn handle_peer_channel_message(
    conn: &mut Box<dyn PeerConnectionApi>,
    msg: Either<NetMessage, ConnEvent>,
) -> Result<(), TransportError> {
    match msg {
        Left(msg) => {
            tracing::debug!(to=%conn.remote_addr() ,"Sending message to peer. Msg: {msg}");
            if let Err(error) = conn.send_message(msg).await {
                tracing::error!(
                    to = %conn.remote_addr(),
                    ?error,
                    "[CONN_LIFECYCLE] Failed to send message to peer"
                );
                return Err(error);
            }
            tracing::debug!(
                to = %conn.remote_addr(),
                "[CONN_LIFECYCLE] Message enqueued on transport socket"
            );
        }
        Right(action) => {
            tracing::debug!(to=%conn.remote_addr(), "Received action from channel");
            match action {
                ConnEvent::NodeAction(NodeEvent::DropConnection(peer)) => {
                    tracing::info!(
                        to = %conn.remote_addr(),
                        peer = %peer,
                        "[CONN_LIFECYCLE] Closing connection per DropConnection action"
                    );
                    return Err(TransportError::ConnectionClosed(conn.remote_addr()));
                }
                ConnEvent::ClosedChannel(reason) => {
                    tracing::info!(
                        to = %conn.remote_addr(),
                        reason = ?reason,
                        "[CONN_LIFECYCLE] Closing connection due to ClosedChannel action"
                    );
                    return Err(TransportError::ConnectionClosed(conn.remote_addr()));
                }
                ConnEvent::StreamSend {
                    stream_id,
                    data,
                    metadata,
                    completion_tx,
                    ..
                } => {
                    tracing::debug!(
                        to = %conn.remote_addr(),
                        stream_id = %stream_id,
                        data_len = data.len(),
                        has_metadata = metadata.is_some(),
                        "[CONN_LIFECYCLE] Sending operations stream data to peer"
                    );
                    if let Err(error) = conn
                        .send_stream_data(stream_id, data, metadata, completion_tx)
                        .await
                    {
                        tracing::error!(
                            to = %conn.remote_addr(),
                            stream_id = %stream_id,
                            ?error,
                            "[CONN_LIFECYCLE] Failed to send stream data to peer"
                        );
                        return Err(error);
                    }
                }
                ConnEvent::PipeStream {
                    outbound_stream_id,
                    inbound_handle,
                    metadata,
                    ..
                } => {
                    tracing::debug!(
                        to = %conn.remote_addr(),
                        stream_id = %outbound_stream_id,
                        total_bytes = inbound_handle.total_bytes(),
                        has_metadata = metadata.is_some(),
                        "[CONN_LIFECYCLE] Piping stream to peer"
                    );
                    if let Err(error) = conn
                        .pipe_stream_data(outbound_stream_id, inbound_handle, metadata)
                        .await
                    {
                        tracing::error!(
                            to = %conn.remote_addr(),
                            stream_id = %outbound_stream_id,
                            ?error,
                            "[CONN_LIFECYCLE] Failed to pipe stream to peer"
                        );
                        return Err(error);
                    }
                }
                other @ ConnEvent::InboundMessage(_)
                | other @ ConnEvent::OutboundMessage(_)
                | other @ ConnEvent::OutboundMessageWithTarget { .. }
                | other @ ConnEvent::NodeAction(_)
                | other @ ConnEvent::TransportClosed { .. } => {
                    unreachable!(
                        "Unexpected action from peer_connection_listener channel: {:?}",
                        other
                    );
                }
            }
        }
    }
    Ok(())
}

async fn notify_transport_closed(
    sender: &Sender<ConnEvent>,
    remote_addr: SocketAddr,
    error: TransportError,
    connection_id: u64,
) {
    if sender
        .send(ConnEvent::TransportClosed {
            remote_addr,
            error,
            connection_id,
        })
        .await
        .is_err()
    {
        tracing::debug!(
            remote = %remote_addr,
            "[CONN_LIFECYCLE] conn_events receiver dropped before handling closure event"
        );
    }
}

/// Drains and sends all pending outbound messages before shutting down.
///
/// This is critical for preventing message loss: when we detect an error condition
/// (transport error, channel closed, etc.), there may be messages that were queued
/// to the channel AFTER we started waiting in select! but BEFORE we detected the error.
/// Without this drain, those messages would be silently lost.
async fn drain_pending_before_shutdown(
    rx: &mut PeerConnChannelRecv,
    conn: &mut Box<dyn PeerConnectionApi>,
    remote_addr: SocketAddr,
) -> usize {
    let mut drained = 0;
    while let Ok(msg) = rx.try_recv() {
        // Best-effort send - connection may already be degraded
        if let Err(e) = handle_peer_channel_message(conn, msg).await {
            tracing::debug!(
                to = %remote_addr,
                ?e,
                drained,
                "[CONN_LIFECYCLE] Error during shutdown drain (expected if connection closed)"
            );
            break;
        }
        drained += 1;
    }
    if drained > 0 {
        tracing::info!(
            to = %remote_addr,
            drained,
            "[CONN_LIFECYCLE] Drained pending messages before shutdown"
        );
    }
    drained
}

/// Listens for messages on a peer connection using drain-then-select pattern.
///
/// On each iteration, drains all pending outbound messages via `try_recv()` before
/// waiting for either new outbound or inbound messages. This approach:
/// 1. Ensures queued outbound messages are sent promptly
/// 2. Avoids starving inbound (which would happen with biased select)
/// 3. No messages are lost due to poll ordering
///
/// IMPORTANT: Before exiting on any error path, we drain pending messages to prevent
/// loss of messages that arrived during the select! wait.
async fn peer_connection_listener(
    mut rx: PeerConnChannelRecv,
    mut conn: Box<dyn PeerConnectionApi>,
    peer_addr: SocketAddr,
    conn_events: Sender<ConnEvent>,
    connection_id: u64,
) {
    let remote_addr = conn.remote_addr();
    tracing::debug!(
        to = %remote_addr,
        %peer_addr,
        "[CONN_LIFECYCLE] Starting peer_connection_listener task"
    );

    loop {
        // Yield to allow the tokio runtime to schedule other tasks.
        // This is critical for cooperative scheduling - without it, tokio's coop
        // budget can be exhausted and channel recv() will return Pending even when
        // messages are available. See https://github.com/tokio-rs/tokio/issues/7108
        //
        // We cannot add this yield inside deterministic_select! because it would
        // break determinism guarantees for simulation testing.
        tokio::task::yield_now().await;

        // Drain pending outbound messages before checking inbound, but with a
        // bounded limit to prevent outbound from starving inbound packet processing.
        // Remaining messages are picked up by the fair select's outbound branch.
        // 8 messages ≈ 0.8ms of sends (8 × ~100µs per encrypt+send).
        const MAX_OUTBOUND_DRAIN: usize = 8;
        let mut drained = 0;
        loop {
            match rx.try_recv() {
                Ok(msg) => {
                    drained += 1;
                    if let Err(error) = handle_peer_channel_message(&mut conn, msg).await {
                        if error.is_transient_send_failure() {
                            tracing::warn!(
                                to = %remote_addr,
                                ?error,
                                "[CONN_LIFECYCLE] Transient send failure, continuing"
                            );
                        } else {
                            tracing::debug!(
                                to = %remote_addr,
                                ?error,
                                "[CONN_LIFECYCLE] Connection closed after channel command"
                            );
                            // Drain any messages that arrived after our try_recv() but before
                            // handle_peer_channel_message returned an error
                            drain_pending_before_shutdown(&mut rx, &mut conn, remote_addr).await;
                            notify_transport_closed(
                                &conn_events,
                                remote_addr,
                                error,
                                connection_id,
                            )
                            .await;
                            return;
                        }
                    }
                    if drained >= MAX_OUTBOUND_DRAIN {
                        break;
                    }
                }
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    tracing::warn!(
                        to = %remote_addr,
                        "[CONN_LIFECYCLE] peer_connection_listener channel closed without explicit DropConnection"
                    );
                    // Channel disconnected means no more messages can arrive
                    notify_transport_closed(
                        &conn_events,
                        remote_addr,
                        TransportError::ConnectionClosed(remote_addr),
                        connection_id,
                    )
                    .await;
                    return;
                }
            }
        }

        // Now wait for either new outbound or inbound messages fairly
        // Uses deterministic_select! for consistent test behavior under DST
        crate::deterministic_select! {
            msg = rx.recv() => {
                match msg {
                    Some(msg) => {
                        if let Err(error) = handle_peer_channel_message(&mut conn, msg).await {
                            if error.is_transient_send_failure() {
                                tracing::warn!(
                                    to = %remote_addr,
                                    ?error,
                                    "[CONN_LIFECYCLE] Transient send failure, continuing"
                                );
                            } else {
                                tracing::debug!(
                                    to = %remote_addr,
                                    ?error,
                                    "[CONN_LIFECYCLE] Connection closed after channel command"
                                );
                                // Drain any messages that arrived while we were processing this one
                                drain_pending_before_shutdown(&mut rx, &mut conn, remote_addr).await;
                                notify_transport_closed(&conn_events, remote_addr, error, connection_id).await;
                                return;
                            }
                        }
                    }
                    None => {
                        tracing::warn!(
                            to = %remote_addr,
                            "[CONN_LIFECYCLE] peer_connection_listener channel closed without explicit DropConnection"
                        );
                        // Channel closed means no more messages can arrive
                        notify_transport_closed(
                            &conn_events,
                            remote_addr,
                            TransportError::ConnectionClosed(remote_addr),
                            connection_id,
                        )
                        .await;
                        return;
                    }
                }
            },
            msg = conn.recv() => {
                match msg {
                    Ok(msg) => match decode_msg(&msg) {
                        Ok(net_message) => {
                            let tx = *net_message.id();
                            tracing::debug!(
                                from = %remote_addr,
                                %tx,
                                tx_type = ?tx.transaction_type(),
                                msg_type = %net_message,
                                "[CONN_LIFECYCLE] Received inbound NetMessage from peer"
                            );
                            if conn_events
                                .send(ConnEvent::InboundMessage(IncomingMessage::with_remote(
                                    net_message,
                                    remote_addr,
                                )))
                                .await
                                .is_err()
                            {
                                tracing::debug!(
                                    from = %remote_addr,
                                    "[CONN_LIFECYCLE] conn_events receiver dropped; stopping listener"
                                );
                                // Drain pending messages - they may still be sendable even if
                                // the conn_events channel is closed
                                drain_pending_before_shutdown(&mut rx, &mut conn, remote_addr).await;
                                return;
                            }
                        }
                        Err(error) => {
                            tracing::error!(
                                from = %remote_addr,
                                ?error,
                                "[CONN_LIFECYCLE] Failed to deserialize inbound message; closing connection"
                            );
                            // Drain pending outbound messages before closing - they may still succeed
                            drain_pending_before_shutdown(&mut rx, &mut conn, remote_addr).await;
                            let transport_error = TransportError::Other(anyhow!(
                                "Failed to deserialize inbound message from {remote_addr}: {error:?}"
                            ));
                            notify_transport_closed(&conn_events, remote_addr, transport_error, connection_id).await;
                            return;
                        }
                    },
                    Err(error) if error.is_transient_send_failure() => {
                        // Transient send failure from internal sends (ACKs, etc.) that
                        // bubbled up through recv(). Don't kill the connection — the idle
                        // timeout is the sole authority on liveness.
                        tracing::warn!(
                            from = %remote_addr,
                            ?error,
                            "[CONN_LIFECYCLE] Transient send failure during recv, continuing"
                        );
                    }
                    Err(error) => {
                        tracing::debug!(
                            from = %remote_addr,
                            ?error,
                            "[CONN_LIFECYCLE] peer_connection_listener terminating after recv error"
                        );
                        // CRITICAL: Drain pending outbound messages before exiting.
                        // Messages may have been queued to the channel while we were
                        // waiting in select!, and they would be lost without this drain.
                        drain_pending_before_shutdown(&mut rx, &mut conn, remote_addr).await;
                        notify_transport_closed(&conn_events, remote_addr, error, connection_id).await;
                        return;
                    }
                }
            },
        }
    }
}

#[inline(always)]
fn decode_msg(data: &[u8]) -> Result<NetMessage, ConnectionError> {
    bincode::deserialize(data).map_err(|err| ConnectionError::Serialization(Some(err)))
}

/// Extract sender information from various message types.
/// Most message types use connection-based routing (sender determined from socket),
/// but ConnectMsg::Request contains the joiner's identity (pub_key) which we need
/// to associate with the transport-layer connection entry.
fn extract_sender_from_message(msg: &NetMessage) -> Option<PeerKeyLocation> {
    match msg {
        NetMessage::V1(msg_v1) => match msg_v1 {
            // ConnectMsg::Request contains the joiner's identity in payload.joiner.
            // The address may be Unknown (peer behind NAT), but the pub_key is always present.
            // We return the joiner so that the transport layer can update its connection entry
            // with the peer's identity, enabling QueryConnections to return the peer.
            NetMessageV1::Connect(ConnectMsg::Request { payload, .. }) => {
                Some(payload.joiner.clone())
            }
            // Other Connect messages (Response, ObservedAddress) and all other operations
            // use hop-by-hop routing via upstream_addr in operation state.
            // No sender/target is embedded - routing is determined by transport layer.
            NetMessageV1::Connect(_)
            | NetMessageV1::Get(_)
            | NetMessageV1::Put(_)
            | NetMessageV1::Update(_)
            | NetMessageV1::Subscribe(_) => None,
            // Other message types don't have sender info
            NetMessageV1::Aborted(_)
            | NetMessageV1::NeighborHosting { .. }
            | NetMessageV1::InterestSync { .. }
            | NetMessageV1::ReadyState { .. } => None,
        },
    }
}

fn extract_sender_from_message_mut(msg: &mut NetMessage) -> Option<&mut PeerKeyLocation> {
    match msg {
        NetMessage::V1(msg_v1) => match msg_v1 {
            // All operations now use hop-by-hop routing via upstream_addr in operation state.
            // No sender/target is embedded in messages - routing is determined by transport layer.
            NetMessageV1::Connect(_)
            | NetMessageV1::Get(_)
            | NetMessageV1::Put(_)
            | NetMessageV1::Update(_)
            | NetMessageV1::Subscribe(_) => None,
            NetMessageV1::Aborted(_)
            | NetMessageV1::NeighborHosting { .. }
            | NetMessageV1::InterestSync { .. }
            | NetMessageV1::ReadyState { .. } => None,
        },
    }
}

// TODO: add testing for the network loop, now it should be possible to do since we don't depend upon having real connections

#[cfg(test)]
mod tests {
    use crate::config::GlobalExecutor;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::sync::mpsc;
    use tokio::time::{Duration, Instant, sleep, timeout};

    /// Regression test for message loss during shutdown.
    ///
    /// This test validates the drain-before-shutdown pattern that prevents message loss
    /// when an error occurs during select! wait. The bug scenario:
    ///
    /// 1. Listener enters select! waiting for inbound/outbound
    /// 2. Message is queued to the channel while select! is waiting
    /// 3. Error occurs (e.g., connection timeout, deserialize failure)
    /// 4. select! returns with the error
    /// 5. BUG: Without drain, we return immediately and the queued message is lost
    /// 6. FIX: Drain pending messages before returning
    ///
    /// This test simulates the pattern by:
    /// - Having a "listener" that waits on select! then encounters an error
    /// - Sending messages during the wait period
    /// - Verifying all messages are processed (drained) before exit
    #[tokio::test]
    async fn test_drain_before_shutdown_prevents_message_loss() {
        let (tx, mut rx) = mpsc::channel::<String>(10);
        let processed = Arc::new(AtomicUsize::new(0));
        let processed_clone = processed.clone();

        // Simulate peer_connection_listener pattern
        let listener = GlobalExecutor::spawn(async move {
            // Drain-then-select loop (simplified)
            loop {
                // Phase 1: Drain pending messages (like the loop at start of peer_connection_listener)
                loop {
                    match rx.try_recv() {
                        Ok(msg) => {
                            tracing::debug!("Drained message: {}", msg);
                            processed_clone.fetch_add(1, Ordering::SeqCst);
                        }
                        Err(mpsc::error::TryRecvError::Empty) => break,
                        Err(mpsc::error::TryRecvError::Disconnected) => return,
                    }
                }

                // Phase 2: Wait for new messages or "error" (simulated by timeout)
                tokio::select! {
                    msg = rx.recv() => {
                        match msg {
                            Some(m) => {
                                tracing::debug!("Received via select: {}", m);
                                processed_clone.fetch_add(1, Ordering::SeqCst);
                            }
                            None => return, // Channel closed
                        }
                    }
                    // Simulate error condition (e.g., connection timeout)
                    _ = sleep(Duration::from_millis(50)) => {
                        tracing::debug!("Simulated error occurred");

                        // CRITICAL: This is the fix - drain before shutdown
                        // Without this, messages queued during the 50ms wait would be lost
                        let mut drained = 0;
                        while let Ok(msg) = rx.try_recv() {
                            tracing::debug!("Drain before shutdown: {}", msg);
                            processed_clone.fetch_add(1, Ordering::SeqCst);
                            drained += 1;
                        }
                        tracing::debug!("Drained {} messages before shutdown", drained);
                        return;
                    }
                }
            }
        });

        // Send messages with timing that causes them to arrive DURING the select! wait
        // This is the race condition that causes message loss without the drain
        GlobalExecutor::spawn(async move {
            // Wait for listener to enter select!
            sleep(Duration::from_millis(10)).await;

            // Send messages while listener is in select! waiting
            for i in 0..5 {
                tx.send(format!("msg{}", i)).await.unwrap();
                sleep(Duration::from_millis(5)).await;
            }
            // Don't close the channel - let the timeout trigger the "error"
        });

        // Wait for listener to complete
        let result = timeout(Duration::from_millis(200), listener).await;
        assert!(result.is_ok(), "Listener should complete");

        // All 5 messages should have been processed
        let count = processed.load(Ordering::SeqCst);
        assert_eq!(
            count, 5,
            "All messages should be processed (drained before shutdown). Got {} instead of 5. \
             If this fails, messages are being lost during the error-exit drain.",
            count
        );
    }

    /// Test that demonstrates what happens WITHOUT the drain (the bug we're preventing).
    /// This test intentionally shows the failure mode when drain is missing.
    ///
    /// The key is to ensure messages are queued AFTER select! starts but the error
    /// fires immediately, so select! returns with error before processing the messages.
    #[tokio::test]
    async fn test_message_loss_without_drain() {
        let (tx, mut rx) = mpsc::channel::<String>(10);
        let processed = Arc::new(AtomicUsize::new(0));
        let processed_clone = processed.clone();

        // Signal when listener is ready for messages
        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<()>();

        // Buggy listener - NO drain before shutdown (demonstrates the bug)
        let listener = GlobalExecutor::spawn(async move {
            // Drain at loop start (the original PR #2255 fix)
            loop {
                match rx.try_recv() {
                    Ok(msg) => {
                        tracing::debug!("Drained message: {}", msg);
                        processed_clone.fetch_add(1, Ordering::SeqCst);
                    }
                    Err(mpsc::error::TryRecvError::Empty) => break,
                    Err(mpsc::error::TryRecvError::Disconnected) => return,
                }
            }

            // Signal that we're about to enter select! and then immediately timeout
            assert!(ready_tx.send(()).is_ok(), "ready_tx receiver dropped");

            // Immediate timeout to simulate error - messages sent after this starts are lost
            tokio::select! {
                biased; // Ensure timeout wins if both are ready
                _ = sleep(Duration::from_millis(1)) => {
                    tracing::debug!("Error occurred - BUG: returning without drain!");
                    // BUG: No drain here! Messages queued during this 1ms window are lost.
                }
                msg = rx.recv() => {
                    if let Some(m) = msg {
                        tracing::debug!("Received via select: {}", m);
                        processed_clone.fetch_add(1, Ordering::SeqCst);
                    }
                }
            }
        });

        // Wait for listener to be ready, then immediately send messages
        let _ready = ready_rx.await;

        // Send messages - these arrive while select! is running with a very short timeout
        // Some or all will be lost because we return without draining
        for i in 0..5 {
            let _sent = tx.send(format!("msg{}", i)).await;
        }

        let _listener_result = timeout(Duration::from_millis(100), listener).await;

        // Without the drain fix, messages are lost
        let count = processed.load(Ordering::SeqCst);
        // The listener exits immediately due to timeout, so no messages should be processed
        assert!(
            count < 5,
            "Without drain, messages should be lost. Got {} (expected < 5). \
             This test validates the bug exists when drain is missing.",
            count
        );
    }

    /// Test that connection_id generation produces unique, monotonically increasing IDs.
    #[test]
    fn test_connection_id_uniqueness() {
        let id1 = super::next_connection_id();
        let id2 = super::next_connection_id();
        let id3 = super::next_connection_id();
        assert!(id2 > id1, "IDs must be monotonically increasing");
        assert!(id3 > id2, "IDs must be monotonically increasing");
    }

    /// Validates the stale TransportClosed detection pattern: when a connection is
    /// replaced (peer reconnected with new identity), a TransportClosed from the old
    /// listener (mismatched connection_id) must not remove the replacement entry.
    #[test]
    fn test_stale_transport_closed_detection() {
        use std::collections::BTreeMap;
        use std::net::SocketAddr;

        let addr: SocketAddr = "1.2.3.4:5678".parse().unwrap();
        let mut connections: BTreeMap<SocketAddr, super::ConnectionEntry> = BTreeMap::new();

        // Insert initial connection (id=10), then replace with new one (id=20)
        let (tx1, _rx1) = mpsc::channel(1);
        connections.insert(
            addr,
            super::ConnectionEntry {
                sender: tx1,
                pub_key: None,
                connection_id: 10,
                created_at: Instant::now(),
            },
        );
        let (tx2, _rx2) = mpsc::channel(1);
        connections.insert(
            addr,
            super::ConnectionEntry {
                sender: tx2,
                pub_key: None,
                connection_id: 20,
                created_at: Instant::now(),
            },
        );

        // Stale TransportClosed (id=10) should not match the current entry (id=20)
        let current = connections.get(&addr).unwrap();
        assert_ne!(current.connection_id, 10, "stale event must not match");
        assert_eq!(current.connection_id, 20);

        // Current TransportClosed (id=20) should match and allow removal
        assert_eq!(current.connection_id, 20, "current event must match");
        connections.remove(&addr);
        assert!(!connections.contains_key(&addr));
    }

    // ============ Zombie detection tests ============

    const TEST_TRANSIENT_TTL: Duration = Duration::from_secs(30);
    // With TTL=30s: zombie_threshold=90s, absolute_zombie_threshold=180s

    #[test]
    fn test_zombie_detection_ignores_young_connections() {
        // Connection younger than zombie threshold (90s) should never be a zombie
        let elapsed = Duration::from_secs(60);
        assert!(
            !super::is_zombie(elapsed, false, false, false, TEST_TRANSIENT_TTL),
            "Young connection should not be zombie"
        );
    }

    #[test]
    fn test_zombie_detection_ignores_ring_connections() {
        // In-ring connection should never be a zombie regardless of age
        let elapsed = Duration::from_secs(400);
        assert!(
            !super::is_zombie(elapsed, true, false, false, TEST_TRANSIENT_TTL),
            "Ring connection should not be zombie"
        );
    }

    #[test]
    fn test_zombie_detection_catches_old_unpromoted() {
        // Old connection that isn't in ring, no pending → zombie
        let elapsed = Duration::from_secs(91); // > 90s
        assert!(
            super::is_zombie(elapsed, false, false, false, TEST_TRANSIENT_TTL),
            "Old unpromoted connection should be zombie"
        );
    }

    #[test]
    fn test_zombie_detection_ignores_pending_reservation() {
        // Old connection not in ring, but has pending reservation → not zombie (under absolute)
        let elapsed = Duration::from_secs(120); // > 90s but < 180s
        assert!(
            !super::is_zombie(elapsed, false, true, false, TEST_TRANSIENT_TTL),
            "Connection with pending reservation below absolute threshold should not be zombie"
        );
    }

    #[test]
    fn test_zombie_absolute_threshold_overrides_pending() {
        // Connection 181s old, has_pending=true, not in ring → IS zombie
        // The absolute threshold (180s) overrides has_pending
        let elapsed = Duration::from_secs(181);
        assert!(
            super::is_zombie(elapsed, false, true, false, TEST_TRANSIENT_TTL),
            "Absolute threshold should override has_pending"
        );
    }

    #[test]
    fn test_zombie_absolute_threshold_spares_ring() {
        // Connection 181s old, in_ring=true → NOT zombie
        let elapsed = Duration::from_secs(181);
        assert!(
            !super::is_zombie(elapsed, true, true, false, TEST_TRANSIENT_TTL),
            "Ring connection should never be zombie even past absolute threshold"
        );
    }

    #[test]
    fn test_zombie_boundary_exactly_at_threshold() {
        // Exactly 90s, no pending, not in ring → NOT zombie (uses > not >=)
        assert!(!super::is_zombie(
            Duration::from_secs(90),
            false,
            false,
            false,
            TEST_TRANSIENT_TTL
        ));
    }

    #[test]
    fn test_zombie_boundary_exactly_at_absolute() {
        // Exactly 180s, has_pending, not in ring → NOT zombie (uses > not >=)
        assert!(!super::is_zombie(
            Duration::from_secs(180),
            false,
            true,
            false,
            TEST_TRANSIENT_TTL
        ));
    }

    #[test]
    fn test_zombie_detection_ignores_gateway_connections() {
        // Gateway connections are intentionally transient and should not be
        // classified as zombies within the 1-hour gateway exemption (#3595).
        let elapsed = Duration::from_secs(400); // Well past normal absolute threshold (180s)
        assert!(
            !super::is_zombie(elapsed, false, false, true, TEST_TRANSIENT_TTL),
            "Gateway connection should not be zombie within exemption window"
        );

        // But truly stale gateway connections (>1 hour) ARE cleaned up.
        let stale = Duration::from_secs(3601);
        assert!(
            super::is_zombie(stale, false, false, true, TEST_TRANSIENT_TTL),
            "Gateway connection past 1-hour cap should be zombie"
        );
    }

    #[test]
    fn test_zombie_thresholds_scale_with_ttl() {
        // With larger TTL (120s, as used in tests), thresholds scale:
        // zombie_threshold=360s, absolute=720s
        let large_ttl = Duration::from_secs(120);
        // 200s: not zombie even without pending (< 360s)
        assert!(!super::is_zombie(
            Duration::from_secs(200),
            false,
            false,
            false,
            large_ttl
        ));
        // 400s: zombie without pending (> 360s)
        assert!(super::is_zombie(
            Duration::from_secs(400),
            false,
            false,
            false,
            large_ttl
        ));
        // 400s with pending: not zombie (< 720s)
        assert!(!super::is_zombie(
            Duration::from_secs(400),
            false,
            true,
            false,
            large_ttl
        ));
        // 721s: zombie even with pending (> 720s)
        assert!(super::is_zombie(
            Duration::from_secs(721),
            false,
            true,
            false,
            large_ttl
        ));
    }
}