transports: Fix missing Poll::Ready(None) event from listenener (#285)

This PR handles the following: - Align Webscoket listener with TCP listener to not miss out on `Poll::Ready(None)` events and enhance logging - Add warnings whenever pending connection IDs do not exist - ~~Filtering out `Poll::Ready` events into `Poll::Pending` without calling the context waker will result in the scheduler not polling in the future. This was happening, for example, for both TCP and Websocket when receiving a connection established event that was previously canceled~~ cc @paritytech/networking Discovered during: #282 --------- Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
paritytech · Nov 12, 2024 · cbc657d · cbc657d
1 parent b1866ae
commit cbc657d
Show file tree

Hide file tree

Showing 2 changed files with 85 additions and 19 deletions.
diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs
@@ -378,20 +378,34 @@ impl Transport for TcpTransport {
     }
 
     fn accept_pending(&mut self, connection_id: ConnectionId) -> crate::Result<()> {
-        let pending = self
-            .pending_inbound_connections
-            .remove(&connection_id)
-            .ok_or(Error::ConnectionDoesntExist(connection_id))?;
+        let pending = self.pending_inbound_connections.remove(&connection_id).ok_or_else(|| {
+            tracing::error!(
+                target: LOG_TARGET,
+                ?connection_id,
+                "Cannot accept non existent pending connection",
+            );
+
+            Error::ConnectionDoesntExist(connection_id)
+        })?;
 
         self.on_inbound_connection(connection_id, pending.connection, pending.address);
 
         Ok(())
     }
 
     fn reject_pending(&mut self, connection_id: ConnectionId) -> crate::Result<()> {
-        self.pending_inbound_connections
-            .remove(&connection_id)
-            .map_or(Err(Error::ConnectionDoesntExist(connection_id)), |_| Ok(()))
+        self.pending_inbound_connections.remove(&connection_id).map_or_else(
+            || {
+                tracing::error!(
+                    target: LOG_TARGET,
+                    ?connection_id,
+                    "Cannot reject non existent pending connection",
+                );
+
+                Err(Error::ConnectionDoesntExist(connection_id))
+            },
+            |_| Ok(()),
+        )
     }
 
     fn reject(&mut self, connection_id: ConnectionId) -> crate::Result<()> {
@@ -529,7 +543,23 @@ impl Stream for TcpTransport {
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         if let Poll::Ready(event) = self.listener.poll_next_unpin(cx) {
             return match event {
-                None | Some(Err(_)) => Poll::Ready(None),
+                None => {
+                    tracing::error!(
+                        target: LOG_TARGET,
+                        "TCP listener terminated, ignore if the node is stopping",
+                    );
+
+                    Poll::Ready(None)
+                }
+                Some(Err(error)) => {
+                    tracing::error!(
+                        target: LOG_TARGET,
+                        ?error,
+                        "TCP listener terminated with error",
+                    );
+
+                    Poll::Ready(None)
+                }
                 Some(Ok((connection, address))) => {
                     let connection_id = self.context.next_connection_id();
                     tracing::trace!(

diff --git a/src/transport/websocket/mod.rs b/src/transport/websocket/mod.rs
@@ -429,20 +429,34 @@ impl Transport for WebSocketTransport {
     }
 
     fn accept_pending(&mut self, connection_id: ConnectionId) -> crate::Result<()> {
-        let pending = self
-            .pending_inbound_connections
-            .remove(&connection_id)
-            .ok_or(Error::ConnectionDoesntExist(connection_id))?;
+        let pending = self.pending_inbound_connections.remove(&connection_id).ok_or_else(|| {
+            tracing::error!(
+                target: LOG_TARGET,
+                ?connection_id,
+                "Cannot accept non existent pending connection",
+            );
+
+            Error::ConnectionDoesntExist(connection_id)
+        })?;
 
         self.on_inbound_connection(connection_id, pending.connection, pending.address);
 
         Ok(())
     }
 
     fn reject_pending(&mut self, connection_id: ConnectionId) -> crate::Result<()> {
-        self.pending_open
-            .remove(&connection_id)
-            .map_or(Err(Error::ConnectionDoesntExist(connection_id)), |_| Ok(()))
+        self.pending_inbound_connections.remove(&connection_id).map_or_else(
+            || {
+                tracing::error!(
+                    target: LOG_TARGET,
+                    ?connection_id,
+                    "Cannot reject non existent pending connection",
+                );
+
+                Err(Error::ConnectionDoesntExist(connection_id))
+            },
+            |_| Ok(()),
+        )
     }
 
     fn open(
@@ -573,11 +587,33 @@ impl Stream for WebSocketTransport {
     type Item = TransportEvent;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        if let Poll::Ready(Some(connection)) = self.listener.poll_next_unpin(cx) {
-            return match connection {
-                Err(_) => Poll::Ready(None),
-                Ok((connection, address)) => {
+        if let Poll::Ready(event) = self.listener.poll_next_unpin(cx) {
+            return match event {
+                None => {
+                    tracing::error!(
+                        target: LOG_TARGET,
+                        "Websocket listener terminated, ignore if the node is stopping",
+                    );
+
+                    Poll::Ready(None)
+                }
+                Some(Err(error)) => {
+                    tracing::error!(
+                        target: LOG_TARGET,
+                        ?error,
+                        "Websocket listener terminated with error",
+                    );
+
+                    Poll::Ready(None)
+                }
+                Some(Ok((connection, address))) => {
                     let connection_id = self.context.next_connection_id();
+                    tracing::trace!(
+                        target: LOG_TARGET,
+                        ?connection_id,
+                        ?address,
+                        "pending inbound Websocket connection",
+                    );
 
                     self.pending_inbound_connections.insert(
                         connection_id,