diff --git a/protocols/request-response/src/lib.rs b/protocols/request-response/src/lib.rs index 86ad6f3472e..a435cbf368f 100644 --- a/protocols/request-response/src/lib.rs +++ b/protocols/request-response/src/lib.rs @@ -88,9 +88,9 @@ pub use handler::ProtocolSupport; use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ - ConnectionDenied, ConnectionHandler, ConnectionId, DialError, NetworkBehaviour, NotifyHandler, - PeerAddresses, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, - behaviour::{AddressChange, ConnectionClosed, DialFailure, FromSwarm}, + ConnectionDenied, ConnectionId, DialError, NetworkBehaviour, NotifyHandler, PeerAddresses, + THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, + behaviour::{AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm}, dial_opts::DialOpts, }; use smallvec::SmallVec; @@ -734,27 +734,35 @@ where } } - /// Preloads a new [`Handler`] with requests that are - /// waiting to be sent to the newly connected peer. - fn preload_new_handler( - &mut self, - handler: &mut Handler, - peer: PeerId, - connection_id: ConnectionId, - remote_address: Option, - ) { + /// Records the connection and flushes its queued requests. Runs on `ConnectionEstablished`. + fn on_connection_established(&mut self, established: ConnectionEstablished) { + let ConnectionEstablished { + peer_id, + connection_id, + endpoint, + .. + } = established; + + let remote_address = match endpoint { + ConnectedPoint::Dialer { address, .. } => Some(address.clone()), + ConnectedPoint::Listener { .. } => None, + }; let mut connection = Connection::new(connection_id, remote_address); - if let Some(pending_requests) = self.pending_outbound_requests.remove(&peer) { + if let Some(pending_requests) = self.pending_outbound_requests.remove(&peer_id) { for request in pending_requests { connection .pending_outbound_responses .insert(request.request_id); - handler.on_behaviour_event(request); + self.pending_events.push_back(ToSwarm::NotifyHandler { + peer_id, + handler: NotifyHandler::One(connection_id), + event: request, + }); } } - self.connected.entry(peer).or_default().push(connection); + self.connected.entry(peer_id).or_default().push(connection); } } @@ -767,12 +775,12 @@ where fn handle_established_inbound_connection( &mut self, - connection_id: ConnectionId, - peer: PeerId, + _: ConnectionId, + _: PeerId, _: &Multiaddr, _: &Multiaddr, ) -> Result, ConnectionDenied> { - let mut handler = Handler::new( + let handler = Handler::new( self.inbound_protocols.clone(), self.codec.clone(), self.config.request_timeout, @@ -780,8 +788,6 @@ where self.config.max_concurrent_streams, ); - self.preload_new_handler(&mut handler, peer, connection_id, None); - Ok(handler) } @@ -809,13 +815,13 @@ where fn handle_established_outbound_connection( &mut self, - connection_id: ConnectionId, - peer: PeerId, - remote_address: &Multiaddr, + _: ConnectionId, + _: PeerId, + _: &Multiaddr, _: Endpoint, _: PortUse, ) -> Result, ConnectionDenied> { - let mut handler = Handler::new( + let handler = Handler::new( self.inbound_protocols.clone(), self.codec.clone(), self.config.request_timeout, @@ -823,20 +829,15 @@ where self.config.max_concurrent_streams, ); - self.preload_new_handler( - &mut handler, - peer, - connection_id, - Some(remote_address.clone()), - ); - Ok(handler) } fn on_swarm_event(&mut self, event: FromSwarm) { self.addresses.on_swarm_event(&event); match event { - FromSwarm::ConnectionEstablished(_) => {} + FromSwarm::ConnectionEstablished(established) => { + self.on_connection_established(established) + } FromSwarm::ConnectionClosed(connection_closed) => { self.on_connection_closed(connection_closed) } diff --git a/protocols/request-response/tests/connection_tracking.rs b/protocols/request-response/tests/connection_tracking.rs new file mode 100644 index 00000000000..d3dd5e5796c --- /dev/null +++ b/protocols/request-response/tests/connection_tracking.rs @@ -0,0 +1,174 @@ +// Copyright 2026 LimeChain +// +// Permission is hereby granted, free of charge, to any person obtaining a +// copy of this software and associated documentation files (the "Software"), +// to deal in the Software without restriction, including without limitation +// the rights to use, copy, modify, merge, publish, distribute, sublicense, +// and/or sell copies of the Software, and to permit persons to whom the +// Software is furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS +// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// DEALINGS IN THE SOFTWARE. + +//! Regression test for the `on_connection_closed` connection-tracking panic +//! (`assertion left == right failed, left: false, right: true`). +//! +//! `Behaviour` records a connection in its `connected` map inside +//! `handle_established_*_connection` (handler-creation time) but ignores +//! `FromSwarm::ConnectionEstablished`. In a *composed* behaviour the swarm calls every sibling's +//! `handle_established_*` in sequence; if a later sibling returns `ConnectionDenied`, this +//! behaviour has already recorded the connection, yet it never enters the swarm's connection pool +//! -- so it is never counted in `remaining_established` and never produces its own +//! `ConnectionClosed`. It lingers as a phantom entry. When the last *real* connection to the peer +//! closes with `remaining_established == 0`, the phantom is still tracked, which used to trip a +//! `debug_assert_eq!(connections.is_empty(), remaining_established == 0)` and panic the node. + +use std::{ + io, + task::{Context, Poll}, +}; + +use futures::prelude::*; +use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, transport::PortUse}; +use libp2p_identity::PeerId; +use libp2p_request_response as request_response; +use libp2p_request_response::{Codec, ProtocolSupport}; +use libp2p_swarm::{ + ConnectionId, NetworkBehaviour, StreamProtocol, ToSwarm, + behaviour::{ConnectionClosed, ConnectionEstablished, FromSwarm}, +}; + +#[derive(Clone, Default)] +struct DummyCodec; + +impl Codec for DummyCodec { + type Protocol = StreamProtocol; + type Request = (); + type Response = (); + + async fn read_request(&mut self, _: &Self::Protocol, _: &mut T) -> io::Result<()> + where + T: AsyncRead + Unpin + Send, + { + Ok(()) + } + async fn read_response(&mut self, _: &Self::Protocol, _: &mut T) -> io::Result<()> + where + T: AsyncRead + Unpin + Send, + { + Ok(()) + } + async fn write_request(&mut self, _: &Self::Protocol, _: &mut T, _: ()) -> io::Result<()> + where + T: AsyncWrite + Unpin + Send, + { + Ok(()) + } + async fn write_response(&mut self, _: &Self::Protocol, _: &mut T, _: ()) -> io::Result<()> + where + T: AsyncWrite + Unpin + Send, + { + Ok(()) + } +} + +fn behaviour() -> request_response::Behaviour { + request_response::Behaviour::with_codec( + DummyCodec, + std::iter::once((StreamProtocol::new("/test/1"), ProtocolSupport::Full)), + request_response::Config::default(), + ) +} + +fn dialer_endpoint() -> ConnectedPoint { + ConnectedPoint::Dialer { + address: "/ip4/127.0.0.1/tcp/1".parse().unwrap(), + role_override: Endpoint::Dialer, + port_use: PortUse::Reuse, + } +} + +/// A connection the swarm hands to `handle_established_*` but that a sibling behaviour then denies +/// never yields a `FromSwarm::ConnectionEstablished`, so it must never be recorded. When the one +/// real connection then closes with `remaining_established == 0`, tracking must be exactly empty -- +/// no phantom left behind. +/// +/// Before the fix (which recorded at `handle_established_*`) the phantom lingered and closing the +/// real connection tripped `debug_assert_eq!(connections.is_empty(), remaining_established == 0)`. +#[test] +fn denied_connection_leaves_no_phantom_on_last_close() { + let mut behaviour = behaviour(); + let peer = PeerId::random(); + let addr: Multiaddr = "/ip4/127.0.0.1/tcp/10000".parse().unwrap(); + let endpoint = dialer_endpoint(); + + let phantom = ConnectionId::new_unchecked(1); + let real = ConnectionId::new_unchecked(2); + + // The swarm speculatively asks every behaviour for a handler before admitting the connection. + // `phantom` is the one a sibling denies: it never enters the pool, so no + // `ConnectionEstablished` (nor `ConnectionClosed`) ever follows for it. + behaviour + .handle_established_inbound_connection(phantom, peer, &addr, &addr) + .unwrap(); + behaviour + .handle_established_inbound_connection(real, peer, &addr, &addr) + .unwrap(); + + // Only the real connection is admitted to the pool. + behaviour.on_swarm_event(FromSwarm::ConnectionEstablished(ConnectionEstablished { + peer_id: peer, + connection_id: real, + endpoint: &endpoint, + failed_addresses: &[], + other_established: 0, + })); + + // It then closes as the last established connection. With no phantom recorded, tracking is + // exactly empty and the invariant holds -- must not panic. + behaviour.on_swarm_event(FromSwarm::ConnectionClosed(ConnectionClosed { + peer_id: peer, + connection_id: real, + endpoint: &endpoint, + cause: None, + remaining_established: 0, + })); +} + +/// The correctness point behind the panic: a denied connection must not be recorded, because a +/// phantom entry makes the peer look connected and routes outbound requests to a dead +/// `connection_id` instead of dialing. After only a denied `handle_established_*` (no +/// `ConnectionEstablished`), a `send_request` must dial the peer, not notify a phantom handler. +#[test] +fn denied_connection_is_not_routable() { + let mut behaviour = behaviour(); + let peer = PeerId::random(); + let addr: Multiaddr = "/ip4/127.0.0.1/tcp/1".parse().unwrap(); + + // A connection handed to the behaviour but denied by a sibling: `handle_established` runs, but + // no `ConnectionEstablished` follows. + behaviour + .handle_established_inbound_connection(ConnectionId::new_unchecked(1), peer, &addr, &addr) + .unwrap(); + + behaviour.send_request(&peer, ()); + + let waker = futures::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + match behaviour.poll(&mut cx) { + // Correct: the peer is not connected, so the behaviour dials it. + Poll::Ready(ToSwarm::Dial { .. }) => {} + Poll::Ready(ToSwarm::NotifyHandler { .. }) => { + panic!("request was routed to a phantom connection instead of dialing the peer") + } + _ => panic!("expected the behaviour to dial the peer for the queued request"), + } +}