[request-response] Refine success & error reporting for inbound reque…

…sts. (#1867)

* Refine error reporting for inbound request handling.

At the moment one can neither get confirmation when a
response has been sent on the underlying transport, nor
is one aware of response omissions. The latter was
originally intended as a feature for support of
one-way protocols, which seems like a bad idea in
hindsight. The lack of notification for sent
responses may prohibit implementation of some
request-response protocols that need to ensure
a happens-before relation between sending a
response and a subsequent request, besides uses
for collecting statistics.

Even with these changes, there is no active notification
for failed inbound requests as a result of connections
unexpectedly closing, as is the case for outbound requests.
Instead, for pending inbound requests this scenario
can be identified if necessary by the absense of both
`InboundFailure` and `ResponseSent` events for a particular
previously received request. Interest in this situation is
not expected to be common and would otherwise require
explicitly tracking all inbound requests in the `RequestResponse`
behaviour, which would be a pity. `RequestResponse::send_response`
now also synchronously returns an error if the inbound upgrade
handling the request has been aborted, due to timeout or
closing of the connection, giving more options for graceful
error handling for inbound requests.

As an aside, the `Throttled` wrapper now no longer emits
inbound or outbound error events occurring in the context
of sending credit requests or responses. This is in addition
to not emitting `ResponseSent` events for ACK responses of
credit grants.

* Update protocols/request-response/src/lib.rs

Co-authored-by: Max Inden <[email protected]>

* Address some minor clippy warnings. (#1868)

* Track pending credit request IDs.

In order to avoid emitting events relating to credit grants or acks
on the public API. The public API should only emit events relating
to the actual requests and responses sent by client code.

* Small cleanup

* Cleanup

* Update versions and changelogs.

* Unreleased

Co-authored-by: Max Inden <[email protected]>

CHANGELOG.md

@@ -25,9 +25,11 @@
 
 # Version 0.32.0 [unreleased]
 
+- Update `libp2p-request-response`.
+
 - Update to `libp2p-mdns-0.26`.
 
-- Update `libp2p-websocket`.
+- Update `libp2p-websocket` minimum patch version.
 
 # Version 0.31.2 [2020-12-02]
 

Cargo.toml

@@ -72,7 +72,7 @@ libp2p-noise = { version = "0.27.0", path = "protocols/noise", optional = true }
 libp2p-ping = { version = "0.25.0", path = "protocols/ping", optional = true }
 libp2p-plaintext = { version = "0.25.0", path = "protocols/plaintext", optional = true }
 libp2p-pnet = { version = "0.19.2", path = "protocols/pnet", optional = true }
-libp2p-request-response = { version = "0.6.0", path = "protocols/request-response", optional = true }
+libp2p-request-response = { version = "0.7.0", path = "protocols/request-response", optional = true }
 libp2p-swarm = { version = "0.25.0", path = "swarm" }
 libp2p-uds = { version = "0.25.0", path = "transports/uds", optional = true }
 libp2p-wasm-ext = { version = "0.25.0", path = "transports/wasm-ext", optional = true }

protocols/request-response/CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.7.0 [unreleased]
+
+- Refine emitted events for inbound requests, introducing
+  the `ResponseSent` event and the `ResponseOmission`
+  inbound failures. This effectively removes previous
+  support for one-way protocols without responses.
+  [PR 1867](https://github.com/libp2p/rust-libp2p/pull/1867).
+
 # 0.6.0 [2020-11-25]
 
 - Update `libp2p-swarm` and `libp2p-core`.

protocols/request-response/Cargo.toml

@@ -2,7 +2,7 @@
 name = "libp2p-request-response"
 edition = "2018"
 description = "Generic Request/Response Protocols"
-version = "0.6.0"
+version = "0.7.0"
 authors = ["Parity Technologies <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

protocols/request-response/src/handler.rs

@@ -119,22 +119,29 @@ pub enum RequestResponseHandlerEvent<TCodec>
 where
     TCodec: RequestResponseCodec
 {
-    /// An inbound request.
+    /// A request has been received.
     Request {
         request_id: RequestId,
         request: TCodec::Request,
         sender: oneshot::Sender<TCodec::Response>
     },
-    /// An inbound response.
+    /// A response has been received.
     Response {
         request_id: RequestId,
         response: TCodec::Response
     },
-    /// An outbound upgrade (i.e. request) timed out.
+    /// A response to an inbound request has been sent.
+    ResponseSent(RequestId),
+    /// A response to an inbound request was omitted as a result
+    /// of dropping the response `sender` of an inbound `Request`.
+    ResponseOmission(RequestId),
+    /// An outbound request timed out while sending the request
+    /// or waiting for the response.
     OutboundTimeout(RequestId),
     /// An outbound request failed to negotiate a mutually supported protocol.
     OutboundUnsupportedProtocols(RequestId),
-    /// An inbound request timed out.
+    /// An inbound request timed out while waiting for the request
+    /// or sending the response.
     InboundTimeout(RequestId),
     /// An inbound request failed to negotiate a mutually supported protocol.
     InboundUnsupportedProtocols(RequestId),
@@ -187,9 +194,16 @@ where
 
     fn inject_fully_negotiated_inbound(
         &mut self,
-        (): (),
-        _: RequestId
+        sent: bool,
+        request_id: RequestId
     ) {
+        if sent {
+            self.pending_events.push_back(
+                RequestResponseHandlerEvent::ResponseSent(request_id))
+        } else {
+            self.pending_events.push_back(
+                RequestResponseHandlerEvent::ResponseOmission(request_id))
+        }
     }
 
     fn inject_fully_negotiated_outbound(

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

@@ -93,7 +93,7 @@ impl<TCodec> InboundUpgrade<NegotiatedSubstream> for ResponseProtocol<TCodec>
 where
     TCodec: RequestResponseCodec + Send + 'static,
 {
-    type Output = ();
+    type Output = bool;
     type Error = io::Error;
     type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;
 
@@ -105,10 +105,12 @@ where
                 if let Ok(response) = self.response_receiver.await {
                     let write = self.codec.write_response(&protocol, &mut io, response);
                     write.await?;
+                } else {
+                    return Ok(false)
                 }
             }
             io.close().await?;
-            Ok(())
+            Ok(true)
         }.boxed()
     }
 }

protocols/request-response/src/lib.rs

@@ -46,18 +46,6 @@
 //! For that purpose, [`RequestResponseCodec::Protocol`] is typically
 //! instantiated with a sum type.
 //!
-//! ## One-Way Protocols
-//!
-//! The implementation supports one-way protocols that do not
-//! have responses. In these cases the [`RequestResponseCodec::Response`] can
-//! be defined as `()` and [`RequestResponseCodec::read_response`] as well as
-//! [`RequestResponseCodec::write_response`] given the obvious implementations.
-//! Note that `RequestResponseMessage::Response` will still be emitted,
-//! immediately after the request has been sent, since `RequestResponseCodec::read_response`
-//! will not actually read anything from the given I/O stream.
-//! [`RequestResponse::send_response`] need not be called for one-way protocols,
-//! i.e. the [`ResponseChannel`] may just be dropped.
-//!
 //! ## Limited Protocol Support
 //!
 //! It is possible to only support inbound or outbound requests for
@@ -115,9 +103,11 @@ pub enum RequestResponseMessage<TRequest, TResponse, TChannelResponse = TRespons
         request_id: RequestId,
         /// The request message.
         request: TRequest,
-        /// The sender of the request who is awaiting a response.
+        /// The channel waiting for the response.
         ///
-        /// See [`RequestResponse::send_response`].
+        /// If this channel is dropped instead of being used to send a response
+        /// via [`RequestResponse::send_response`], a [`RequestResponseEvent::InboundFailure`]
+        /// with [`InboundFailure::ResponseOmission`] is emitted.
         channel: ResponseChannel<TChannelResponse>,
     },
     /// A response message.
@@ -151,6 +141,14 @@ pub enum RequestResponseEvent<TRequest, TResponse, TChannelResponse = TResponse>
         error: OutboundFailure,
     },
     /// An inbound request failed.
+    ///
+    /// > **Note**: The case whereby a connection on which a response is sent
+    /// > closes after [`RequestResponse::send_response`] already succeeded
+    /// > but before the response could be sent on the connection is reflected
+    /// > by there being no [`RequestResponseEvent::ResponseSent`] event.
+    /// > Code interested in ensuring a response has been successfully
+    /// > handed to the transport layer, e.g. before continuing with the next
+    /// > step in a multi-step protocol, should listen to these events.
     InboundFailure {
         /// The peer from whom the request was received.
         peer: PeerId,
@@ -159,6 +157,16 @@ pub enum RequestResponseEvent<TRequest, TResponse, TChannelResponse = TResponse>
         /// The error that occurred.
         error: InboundFailure,
     },
+    /// A response to an inbound request has been sent.
+    ///
+    /// When this event is received, the response has been flushed on
+    /// the underlying transport connection.
+    ResponseSent {
+        /// The peer to whom the response was sent.
+        peer: PeerId,
+        /// The ID of the inbound request whose response was sent.
+        request_id: RequestId,
+    },
 }
 
 /// Possible failures occurring in the context of sending
@@ -186,14 +194,17 @@ pub enum OutboundFailure {
 #[derive(Debug)]
 pub enum InboundFailure {
     /// The inbound request timed out, either while reading the
-    /// incoming request or before a response is sent, i.e. if
+    /// incoming request or before a response is sent, e.g. if
     /// [`RequestResponse::send_response`] is not called in a
     /// timely manner.
     Timeout,
-    /// The local peer supports none of the requested protocols.
+    /// The local peer supports none of the protocols requested
+    /// by the remote.
     UnsupportedProtocols,
-    /// The connection closed before a response was delivered.
-    ConnectionClosed,
+    /// The local peer failed to respond to an inbound request
+    /// due to the [`ResponseChannel`] being dropped instead of
+    /// being passed to [`RequestResponse::send_response`].
+    ResponseOmission,
 }
 
 /// A channel for sending a response to an inbound request.
@@ -379,17 +390,18 @@ where
 
     /// Initiates sending a response to an inbound request.
     ///
-    /// If the `ResponseChannel` is already closed due to a timeout,
-    /// the response is discarded and eventually [`RequestResponseEvent::InboundFailure`]
-    /// is emitted by `RequestResponse::poll`.
+    /// If the `ResponseChannel` is already closed due to a timeout or
+    /// the connection being closed, the response is returned as an `Err`
+    /// for further handling. Once the response has been successfully sent
+    /// on the corresponding connection, [`RequestResponseEvent::ResponseSent`]
+    /// is emitted.
     ///
-    /// The provided `ResponseChannel` is obtained from a
+    /// The provided `ResponseChannel` is obtained from an inbound
     /// [`RequestResponseMessage::Request`].
-    pub fn send_response(&mut self, ch: ResponseChannel<TCodec::Response>, rs: TCodec::Response) {
-        // Fails only if the inbound upgrade timed out waiting for the response,
-        // in which case the handler emits `RequestResponseHandlerEvent::InboundTimeout`
-        // which in turn results in `RequestResponseEvent::InboundFailure`.
-        let _ = ch.sender.send(rs);
+    pub fn send_response(&mut self, ch: ResponseChannel<TCodec::Response>, rs: TCodec::Response)
+        -> Result<(), TCodec::Response>
+    {
+        ch.sender.send(rs)
     }
 
     /// Adds a known address for a peer that can be used for
@@ -577,6 +589,20 @@ where
                     RequestResponseEvent::Message { peer, message }
                 ));
             }
+            RequestResponseHandlerEvent::ResponseSent(request_id) => {
+                self.pending_events.push_back(
+                    NetworkBehaviourAction::GenerateEvent(
+                        RequestResponseEvent::ResponseSent { peer, request_id }));
+            }
+            RequestResponseHandlerEvent::ResponseOmission(request_id) => {
+                self.pending_events.push_back(
+                    NetworkBehaviourAction::GenerateEvent(
+                        RequestResponseEvent::InboundFailure {
+                            peer,
+                            request_id,
+                            error: InboundFailure::ResponseOmission
+                        }));
+            }
             RequestResponseHandlerEvent::OutboundTimeout(request_id) => {
                 if let Some((peer, _conn)) = self.pending_responses.remove(&request_id) {
                     self.pending_events.push_back(