Restore RequestResponse::throttled.

protocols/request-response/Cargo.toml

@@ -15,9 +15,9 @@ futures = "0.3.1"
 libp2p-core = { version = "0.21.0", path = "../../core" }
 libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
 log = "0.4.11"
-lru = "0.6"
 rand = "0.7"
 smallvec = "1.4"
+thiserror = "1.0.20"
 wasm-timer = "0.2"
 
 [dev-dependencies]

protocols/request-response/src/handler.rs

@@ -22,6 +22,7 @@ mod protocol;
 
 use crate::{EMPTY_QUEUE_SHRINK_THRESHOLD, RequestId};
 use crate::codec::RequestResponseCodec;
+use protocol::InboundError;
 
 pub use protocol::{RequestProtocol, ResponseProtocol, ProtocolSupport};
 
@@ -47,6 +48,7 @@ use smallvec::SmallVec;
 use std::{
     collections::VecDeque,
     io,
+    sync::{atomic::{AtomicU64, Ordering}, Arc},
     time::Duration,
     task::{Context, Poll}
 };
@@ -79,9 +81,10 @@ where
     /// Inbound upgrades waiting for the incoming request.
     inbound: FuturesUnordered<BoxFuture<'static,
         Result<
-            (TCodec::Request, oneshot::Sender<TCodec::Response>),
+            ((RequestId, TCodec::Request), oneshot::Sender<TCodec::Response>),
             oneshot::Canceled
         >>>,
+    inbound_request_id: Arc<AtomicU64>
 }
 
 impl<TCodec> RequestResponseHandler<TCodec>
@@ -93,6 +96,7 @@ where
         codec: TCodec,
         keep_alive_timeout: Duration,
         substream_timeout: Duration,
+        inbound_request_id: Arc<AtomicU64>
     ) -> Self {
         Self {
             inbound_protocols,
@@ -104,6 +108,7 @@ where
             inbound: FuturesUnordered::new(),
             pending_events: VecDeque::new(),
             pending_error: None,
+            inbound_request_id
         }
     }
 }
@@ -117,6 +122,7 @@ where
 {
     /// An inbound request.
     Request {
+        request_id: RequestId,
         request: TCodec::Request,
         sender: oneshot::Sender<TCodec::Response>
     },
@@ -130,9 +136,11 @@ where
     /// An outbound request failed to negotiate a mutually supported protocol.
     OutboundUnsupportedProtocols(RequestId),
     /// An inbound request timed out.
-    InboundTimeout,
+    InboundTimeout(RequestId),
     /// An inbound request failed to negotiate a mutually supported protocol.
-    InboundUnsupportedProtocols,
+    InboundUnsupportedProtocols(RequestId),
+    /// An inbound request was not answered with a response.
+    InboundResponseOmission(RequestId)
 }
 
 impl<TCodec> ProtocolsHandler for RequestResponseHandler<TCodec>
@@ -145,7 +153,7 @@ where
     type InboundProtocol = ResponseProtocol<TCodec>;
     type OutboundProtocol = RequestProtocol<TCodec>;
     type OutboundOpenInfo = RequestId;
-    type InboundOpenInfo = ();
+    type InboundOpenInfo = RequestId;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         // A channel for notifying the handler when the inbound
@@ -156,6 +164,8 @@ where
         // response is sent.
         let (rs_send, rs_recv) = oneshot::channel();
 
+        let request_id = RequestId(self.inbound_request_id.fetch_add(1, Ordering::Relaxed));
+
         // By keeping all I/O inside the `ResponseProtocol` and thus the
         // inbound substream upgrade via above channels, we ensure that it
         // is all subject to the configured timeout without extra bookkeeping
@@ -167,20 +177,21 @@ where
             codec: self.codec.clone(),
             request_sender: rq_send,
             response_receiver: rs_recv,
+            request_id
         };
 
         // The handler waits for the request to come in. It then emits
         // `RequestResponseHandlerEvent::Request` together with a
         // `ResponseChannel`.
         self.inbound.push(rq_recv.map_ok(move |rq| (rq, rs_send)).boxed());
 
-        SubstreamProtocol::new(proto, ()).with_timeout(self.substream_timeout)
+        SubstreamProtocol::new(proto, request_id).with_timeout(self.substream_timeout)
     }
 
     fn inject_fully_negotiated_inbound(
         &mut self,
         (): (),
-        (): ()
+        _id: RequestId
     ) {
         // Nothing to do, as the response has already been sent
         // as part of the upgrade.
@@ -231,13 +242,12 @@ where
 
     fn inject_listen_upgrade_error(
         &mut self,
-        (): Self::InboundOpenInfo,
-        error: ProtocolsHandlerUpgrErr<io::Error>
+        info: RequestId,
+        error: ProtocolsHandlerUpgrErr<InboundError>
     ) {
         match error {
             ProtocolsHandlerUpgrErr::Timeout => {
-                self.pending_events.push_back(
-                    RequestResponseHandlerEvent::InboundTimeout);
+                self.pending_events.push_back(RequestResponseHandlerEvent::InboundTimeout(info))
             }
             ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
                 // The local peer merely doesn't support the protocol(s) requested.
@@ -246,12 +256,23 @@ where
                 // An event is reported to permit user code to react to the fact that
                 // the local peer does not support the requested protocol(s).
                 self.pending_events.push_back(
-                    RequestResponseHandlerEvent::InboundUnsupportedProtocols);
+                    RequestResponseHandlerEvent::InboundUnsupportedProtocols(info));
             }
-            _ => {
-                // Anything else is considered a fatal error or misbehaviour of
-                // the remote peer and results in closing the connection.
-                self.pending_error = Some(error);
+            ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)) => {
+                self.pending_error = Some(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
+            }
+            ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)) => match e {
+                InboundError::Io(e) => {
+                    let e = UpgradeError::Apply(e);
+                    self.pending_error = Some(ProtocolsHandlerUpgrErr::Upgrade(e))
+                }
+                InboundError::ResponseOmission(id) => {
+                    let e = RequestResponseHandlerEvent::InboundResponseOmission(id);
+                    self.pending_events.push_back(e)
+                }
+            }
+            ProtocolsHandlerUpgrErr::Timer => {
+                self.pending_error = Some(ProtocolsHandlerUpgrErr::Timer)
             }
         }
     }
@@ -282,12 +303,12 @@ where
         // Check for inbound requests.
         while let Poll::Ready(Some(result)) = self.inbound.poll_next_unpin(cx) {
             match result {
-                Ok((rq, rs_sender)) => {
+                Ok(((id, rq), rs_sender)) => {
                     // We received an inbound request.
                     self.keep_alive = KeepAlive::Yes;
                     return Poll::Ready(ProtocolsHandlerEvent::Custom(
                         RequestResponseHandlerEvent::Request {
-                            request: rq, sender: rs_sender
+                            request_id: id, request: rq, sender: rs_sender
                         }))
                 }
                 Err(oneshot::Canceled) => {

protocols/request-response/src/handler/protocol.rs

@@ -61,6 +61,17 @@ impl ProtocolSupport {
     }
 }
 
+/// Possible inbound upgrade errors.
+#[derive(Debug, thiserror::Error)]
+pub enum InboundError {
+    /// Some I/O error occured.
+    #[error("inbound i/o error: {0}")]
+    Io(#[from] io::Error),
+    /// The request was not answered with a response.
+    #[error("no response to request {0:?}")]
+    ResponseOmission(RequestId)
+}
+
 /// Response substream upgrade protocol.
 ///
 /// Receives a request and sends a response.
@@ -71,8 +82,10 @@ where
 {
     pub(crate) codec: TCodec,
     pub(crate) protocols: SmallVec<[TCodec::Protocol; 2]>,
-    pub(crate) request_sender: oneshot::Sender<TCodec::Request>,
-    pub(crate) response_receiver: oneshot::Receiver<TCodec::Response>
+    pub(crate) request_sender: oneshot::Sender<(RequestId, TCodec::Request)>,
+    pub(crate) response_receiver: oneshot::Receiver<TCodec::Response>,
+    pub(crate) request_id: RequestId
+
 }
 
 impl<TCodec> UpgradeInfo for ResponseProtocol<TCodec>
@@ -92,17 +105,20 @@ where
     TCodec: RequestResponseCodec + Send + 'static,
 {
     type Output = ();
-    type Error = io::Error;
+    type Error = InboundError;
     type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
 
     fn upgrade_inbound(mut self, mut io: NegotiatedSubstream, protocol: Self::Info) -> Self::Future {
         async move {
             let read = self.codec.read_request(&protocol, &mut io);
             let request = read.await?;
-            if let Ok(()) = self.request_sender.send(request) {
+            if let Ok(()) = self.request_sender.send((self.request_id, request)) {
                 if let Ok(response) = self.response_receiver.await {
                     let write = self.codec.write_response(&protocol, &mut io, response);
                     write.await?;
+                } else {
+                    io.close().await?;
+                    return Err(InboundError::ResponseOmission(self.request_id))
                 }
             }
             io.close().await?;