fix(request-response): Report failures

examples/file-sharing/src/network.rs

@@ -8,7 +8,7 @@ use libp2p::{
     identity, kad,
     multiaddr::Protocol,
     noise,
-    request_response::{self, ProtocolSupport, RequestId, ResponseChannel},
+    request_response::{self, OutboundRequestId, ProtocolSupport, ResponseChannel},
     swarm::{NetworkBehaviour, Swarm, SwarmEvent},
     tcp, yamux, PeerId,
 };
@@ -175,7 +175,7 @@ pub(crate) struct EventLoop {
     pending_start_providing: HashMap<kad::QueryId, oneshot::Sender<()>>,
     pending_get_providers: HashMap<kad::QueryId, oneshot::Sender<HashSet<PeerId>>>,
     pending_request_file:
-        HashMap<RequestId, oneshot::Sender<Result<Vec<u8>, Box<dyn Error + Send>>>>,
+        HashMap<OutboundRequestId, oneshot::Sender<Result<Vec<u8>, Box<dyn Error + Send>>>>,
 }
 
 impl EventLoop {

protocols/autonat/src/behaviour.rs

@@ -32,7 +32,7 @@ use instant::Instant;
 use libp2p_core::{multiaddr::Protocol, ConnectedPoint, Endpoint, Multiaddr};
 use libp2p_identity::PeerId;
 use libp2p_request_response::{
-    self as request_response, ProtocolSupport, RequestId, ResponseChannel,
+    self as request_response, InboundRequestId, OutboundRequestId, ProtocolSupport, ResponseChannel,
 };
 use libp2p_swarm::{
     behaviour::{
@@ -187,14 +187,14 @@ pub struct Behaviour {
         PeerId,
         (
             ProbeId,
-            RequestId,
+            InboundRequestId,
             Vec<Multiaddr>,
             ResponseChannel<DialResponse>,
         ),
     >,
 
     // Ongoing outbound probes and mapped to the inner request id.
-    ongoing_outbound: HashMap<RequestId, ProbeId>,
+    ongoing_outbound: HashMap<OutboundRequestId, ProbeId>,
 
     // Connected peers with the observed address of each connection.
     // If the endpoint of a connection is relayed or not global (in case of Config::only_global_ips),

protocols/autonat/src/behaviour/as_client.rs

@@ -29,7 +29,7 @@ use futures_timer::Delay;
 use instant::Instant;
 use libp2p_core::Multiaddr;
 use libp2p_identity::PeerId;
-use libp2p_request_response::{self as request_response, OutboundFailure, RequestId};
+use libp2p_request_response::{self as request_response, OutboundFailure, OutboundRequestId};
 use libp2p_swarm::{ConnectionId, ListenAddresses, PollParameters, ToSwarm};
 use rand::{seq::SliceRandom, thread_rng};
 use std::{
@@ -91,7 +91,7 @@ pub(crate) struct AsClient<'a> {
     pub(crate) throttled_servers: &'a mut Vec<(PeerId, Instant)>,
     pub(crate) nat_status: &'a mut NatStatus,
     pub(crate) confidence: &'a mut usize,
-    pub(crate) ongoing_outbound: &'a mut HashMap<RequestId, ProbeId>,
+    pub(crate) ongoing_outbound: &'a mut HashMap<OutboundRequestId, ProbeId>,
     pub(crate) last_probe: &'a mut Option<Instant>,
     pub(crate) schedule_probe: &'a mut Delay,
     pub(crate) listen_addresses: &'a ListenAddresses,
@@ -118,7 +118,7 @@ impl<'a> HandleInnerEvent for AsClient<'a> {
                 let probe_id = self
                     .ongoing_outbound
                     .remove(&request_id)
-                    .expect("RequestId exists.");
+                    .expect("OutboundRequestId exists.");
 
                 let event = match response.result.clone() {
                     Ok(address) => OutboundProbeEvent::Response {

protocols/autonat/src/behaviour/as_server.rs

@@ -26,7 +26,7 @@ use instant::Instant;
 use libp2p_core::{multiaddr::Protocol, Multiaddr};
 use libp2p_identity::PeerId;
 use libp2p_request_response::{
-    self as request_response, InboundFailure, RequestId, ResponseChannel,
+    self as request_response, InboundFailure, InboundRequestId, ResponseChannel,
 };
 use libp2p_swarm::{
     dial_opts::{DialOpts, PeerCondition},
@@ -85,7 +85,7 @@ pub(crate) struct AsServer<'a> {
         PeerId,
         (
             ProbeId,
-            RequestId,
+            InboundRequestId,
             Vec<Multiaddr>,
             ResponseChannel<DialResponse>,
         ),

protocols/autonat/tests/test_client.rs

@@ -61,7 +61,7 @@ async fn test_auto_probe() {
     match client.next_behaviour_event().await {
         Event::OutboundProbe(OutboundProbeEvent::Error { peer, error, .. }) => {
             assert!(peer.is_none());
-            assert_eq!(error, OutboundProbeError::NoAddresses);
+            assert!(matches!(error, OutboundProbeError::NoAddresses));
         }
         other => panic!("Unexpected behaviour event: {other:?}."),
     }
@@ -181,10 +181,10 @@ async fn test_confidence() {
                         peer,
                         error,
                     } if !test_public => {
-                        assert_eq!(
+                        assert!(matches!(
                             error,
                             OutboundProbeError::Response(ResponseError::DialError)
-                        );
+                        ));
                         (peer.unwrap(), probe_id)
                     }
                     other => panic!("Unexpected Outbound Event: {other:?}"),
@@ -261,7 +261,7 @@ async fn test_throttle_server_period() {
     match client.next_behaviour_event().await {
         Event::OutboundProbe(OutboundProbeEvent::Error { peer, error, .. }) => {
             assert!(peer.is_none());
-            assert_eq!(error, OutboundProbeError::NoServer);
+            assert!(matches!(error, OutboundProbeError::NoServer));
         }
         other => panic!("Unexpected behaviour event: {other:?}."),
     }

protocols/autonat/tests/test_server.rs

@@ -168,7 +168,10 @@ async fn test_dial_error() {
         }) => {
             assert_eq!(probe_id, request_probe_id);
             assert_eq!(peer, client_id);
-            assert_eq!(error, InboundProbeError::Response(ResponseError::DialError));
+            assert!(matches!(
+                error,
+                InboundProbeError::Response(ResponseError::DialError)
+            ));
         }
         other => panic!("Unexpected behaviour event: {other:?}."),
     }
@@ -252,10 +255,10 @@ async fn test_throttle_peer_max() {
         }) => {
             assert_eq!(client_id, peer);
             assert_ne!(first_probe_id, probe_id);
-            assert_eq!(
+            assert!(matches!(
                 error,
                 InboundProbeError::Response(ResponseError::DialRefused)
-            )
+            ));
         }
         other => panic!("Unexpected behaviour event: {other:?}."),
     };

protocols/perf/src/client.rs

@@ -37,10 +37,10 @@ use crate::{protocol::Response, RunDuration, RunParams};
 
 /// Connection identifier.
 #[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)]
-pub struct RunId(request_response::RequestId);
+pub struct RunId(request_response::OutboundRequestId);
 
-impl From<request_response::RequestId> for RunId {
-    fn from(value: request_response::RequestId) -> Self {
+impl From<request_response::OutboundRequestId> for RunId {
+    fn from(value: request_response::OutboundRequestId) -> Self {
         Self(value)
     }
 }

protocols/rendezvous/src/client.rs

@@ -26,7 +26,7 @@ use futures::stream::FuturesUnordered;
 use futures::stream::StreamExt;
 use libp2p_core::{Endpoint, Multiaddr, PeerRecord};
 use libp2p_identity::{Keypair, PeerId, SigningError};
-use libp2p_request_response::{ProtocolSupport, RequestId};
+use libp2p_request_response::{OutboundRequestId, ProtocolSupport};
 use libp2p_swarm::{
     ConnectionDenied, ConnectionId, ExternalAddresses, FromSwarm, NetworkBehaviour, PollParameters,
     THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
@@ -41,8 +41,8 @@ pub struct Behaviour {
 
     keypair: Keypair,
 
-    waiting_for_register: HashMap<RequestId, (PeerId, Namespace)>,
-    waiting_for_discovery: HashMap<RequestId, (PeerId, Option<Namespace>)>,
+    waiting_for_register: HashMap<OutboundRequestId, (PeerId, Namespace)>,
+    waiting_for_discovery: HashMap<OutboundRequestId, (PeerId, Option<Namespace>)>,
 
     /// Hold addresses of all peers that we have discovered so far.
     ///
@@ -337,7 +337,7 @@ impl NetworkBehaviour for Behaviour {
 }
 
 impl Behaviour {
-    fn event_for_outbound_failure(&mut self, req_id: &RequestId) -> Option<Event> {
+    fn event_for_outbound_failure(&mut self, req_id: &OutboundRequestId) -> Option<Event> {
         if let Some((rendezvous_node, namespace)) = self.waiting_for_register.remove(req_id) {
             return Some(Event::RegisterFailed {
                 rendezvous_node,
@@ -357,7 +357,11 @@ impl Behaviour {
         None
     }
 
-    fn handle_response(&mut self, request_id: &RequestId, response: Message) -> Option<Event> {
+    fn handle_response(
+        &mut self,
+        request_id: &OutboundRequestId,
+        response: Message,
+    ) -> Option<Event> {
         match response {
             RegisterResponse(Ok(ttl)) => {
                 if let Some((rendezvous_node, namespace)) =

protocols/request-response/src/handler.rs

@@ -24,7 +24,7 @@
 
 use crate::codec::Codec;
 use crate::handler::protocol::Protocol;
-use crate::{RequestId, EMPTY_QUEUE_SHRINK_THRESHOLD};
+use crate::{InboundRequestId, OutboundRequestId, EMPTY_QUEUE_SHRINK_THRESHOLD};
 
 use futures::channel::mpsc;
 use futures::{channel::oneshot, prelude::*};
@@ -67,13 +67,13 @@
     requested_outbound: VecDeque<OutboundMessage<TCodec>>,
     /// A channel for receiving inbound requests.
     inbound_receiver: mpsc::Receiver<(
-        RequestId,
+        InboundRequestId,
         TCodec::Request,
         oneshot::Sender<TCodec::Response>,
     )>,
     /// The [`mpsc::Sender`] for the above receiver. Cloned for each inbound request.
     inbound_sender: mpsc::Sender<(
-        RequestId,
+        InboundRequestId,
         TCodec::Request,
         oneshot::Sender<TCodec::Response>,
     )>,
@@ -83,6 +83,12 @@
     worker_streams: futures_bounded::FuturesMap<RequestId, Result<Event<TCodec>, io::Error>>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+enum RequestId {
+    Inbound(InboundRequestId),
+    Outbound(OutboundRequestId),
+}
+
 impl<TCodec> Handler<TCodec>
 where
     TCodec: Codec + Send + Clone + 'static,
@@ -112,6 +118,11 @@
         }
     }
 
+    /// Returns the next inbound request ID.
+    fn next_inbound_request_id(&mut self) -> InboundRequestId {
+        InboundRequestId(self.inbound_request_id.fetch_add(1, Ordering::Relaxed))
+    }
+
     fn on_fully_negotiated_inbound(
         &mut self,
         FullyNegotiatedInbound {
@@ -123,7 +134,7 @@
         >,
     ) {
         let mut codec = self.codec.clone();
-        let request_id = RequestId(self.inbound_request_id.fetch_add(1, Ordering::Relaxed));
+        let request_id = self.next_inbound_request_id();
         let mut sender = self.inbound_sender.clone();
 
         let recv = async move {
@@ -153,7 +164,7 @@
 
         if self
             .worker_streams
-            .try_push(request_id, recv.boxed())
+            .try_push(RequestId::Inbound(request_id), recv.boxed())
             .is_err()
         {
             log::warn!("Dropping inbound stream because we are at capacity")
@@ -193,7 +204,7 @@
 
         if self
             .worker_streams
-            .try_push(request_id, send.boxed())
+            .try_push(RequestId::Outbound(request_id), send.boxed())
             .is_err()
         {
             log::warn!("Dropping outbound stream because we are at capacity")
@@ -254,31 +265,34 @@
 {
     /// A request has been received.
     Request {
-        request_id: RequestId,
+        request_id: InboundRequestId,
         request: TCodec::Request,
         sender: oneshot::Sender<TCodec::Response>,
     },
     /// A response has been received.
     Response {
-        request_id: RequestId,
+        request_id: OutboundRequestId,
         response: TCodec::Response,
     },
     /// A response to an inbound request has been sent.
-    ResponseSent(RequestId),
+    ResponseSent(InboundRequestId),
     /// A response to an inbound request was omitted as a result
     /// of dropping the response `sender` of an inbound `Request`.
-    ResponseOmission(RequestId),
+    ResponseOmission(InboundRequestId),
     /// An outbound request timed out while sending the request
     /// or waiting for the response.
-    OutboundTimeout(RequestId),
+    OutboundTimeout(OutboundRequestId),
     /// An outbound request failed to negotiate a mutually supported protocol.
-    OutboundUnsupportedProtocols(RequestId),
+    OutboundUnsupportedProtocols(OutboundRequestId),
     OutboundStreamFailed {
-        request_id: RequestId,
+        request_id: OutboundRequestId,
         error: io::Error,
     },
+    /// An inbound request timed out while waiting for the request
+    /// or sending the response.
+    InboundTimeout(InboundRequestId),
     InboundStreamFailed {
-        request_id: RequestId,
+        request_id: InboundRequestId,
         error: io::Error,
     },
 }
@@ -322,6 +336,10 @@
                 .field("request_id", &request_id)
                 .field("error", &error)
                 .finish(),
+            Event::InboundTimeout(request_id) => f
+                .debug_tuple("Event::InboundTimeout")
+                .field(request_id)
+                .finish(),
             Event::InboundStreamFailed { request_id, error } => f
                 .debug_struct("Event::InboundStreamFailed")
                 .field("request_id", &request_id)
@@ -332,7 +350,7 @@
 }
 
 pub struct OutboundMessage<TCodec: Codec> {
-    pub(crate) request_id: RequestId,
+    pub(crate) request_id: OutboundRequestId,
     pub(crate) request: TCodec::Request,
     pub(crate) protocols: SmallVec<[TCodec::Protocol; 2]>,
 }
@@ -381,22 +399,50 @@
         cx: &mut Context<'_>,
     ) -> Poll<ConnectionHandlerEvent<Protocol<TCodec::Protocol>, (), Self::ToBehaviour, Self::Error>>
     {
         loop {
             match self.worker_streams.poll_unpin(cx) {
                 Poll::Ready((_, Ok(Ok(event)))) => {
-                    return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event))
+                    return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event));
                 }
                 Poll::Ready((id, Ok(Err(e)))) => {
-                    log::debug!("Stream for request {id} failed: {e}");
+                    let event = match id {
+                        RequestId::Inbound(id) => {
+                            log::debug!("Stream for inbound request {id} failed: {e}");
+                            Event::InboundStreamFailed {
+                                request_id: id,
+                                error: e,
+                            }
+                        }
+                        RequestId::Outbound(id) => {
+                            log::debug!("Stream for outbound request {id} failed: {e}");
+                            Event::OutboundStreamFailed {
+                                request_id: id,
+                                error: e,
+                            }
+                        }
+                    };
+
+                    return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event));
                 }
                 Poll::Ready((id, Err(futures_bounded::Timeout { .. }))) => {
-                    log::debug!("Stream for request {id} timed out");
+                    let event = match id {
+                        RequestId::Inbound(id) => {
+                            log::debug!("Stream for inbound request {id} timed out");
+                            Event::InboundTimeout(id)
+                        }
+                        RequestId::Outbound(id) => {
+                            log::debug!("Stream for outbound request {id} timed out");
+                            Event::OutboundTimeout(id)
+                        }
+                    };
+
+                    return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event));
                 }
                 Poll::Pending => break,
             }
         }
 
-        // Drain pending events.
+        // Drain pending events that were produced by `worker_streams`.
         if let Some(event) = self.pending_events.pop_front() {
             return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event));
         } else if self.pending_events.capacity() > EMPTY_QUEUE_SHRINK_THRESHOLD {