[request-response] Refine success & error reporting for inbound requests.

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(