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/Cargo.toml

@@ -32,6 +32,7 @@ json = ["dep:serde", "dep:serde_json", "libp2p-swarm/macros"]
 cbor = ["dep:serde", "dep:cbor4ii", "libp2p-swarm/macros"]
 
 [dev-dependencies]
+anyhow = "1.0.75"
 async-std = { version = "1.6.2", features = ["attributes"] }
 env_logger = "0.10.0"
 libp2p-noise = { workspace = true }