yamux/control: Ensure poll next inbound is not called after errors (#445)

We have recently fixed a spurious issue that caused the rust-yamux
component to panic.

Litep2p transitioned the internal state of the connection back to Idle.
This causes litep2p to poll the component again. Then rust-yamux would
use the same buffer offset that caused the panic in the first place.

Instead, this PR ensures that litep2p never polls the connection once
yamux encounters an error.
This is aligned with the behavior of transport layers that close the
connection on the first yamux error.

Part of: paritytech/polkadot-sdk#9169

cc @paritytech/networking

---------

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

Author: lexnv

Cargo.toml

@@ -50,7 +50,7 @@ x25519-dalek = "2.0.1"
 x509-parser = "0.17.0"
 yasna = "0.5.0"
 zeroize = "1.8.1"
-yamux = "0.13.5"
+yamux = "0.13.7"
 
 # Websocket related dependencies.
 tokio-tungstenite = { version = "0.27.0", features = ["rustls-tls-native-roots", "url"], optional = true }

src/yamux/control.rs

@@ -19,6 +19,8 @@ use std::{
     task::{Context, Poll},
 };
 
+const LOG_TARGET: &str = "litep2p::yamux::control";
+
 /// A Yamux [`Connection`] controller.
 ///
 /// This presents an alternative API for using a yamux [`Connection`].
@@ -81,7 +83,25 @@ where
                 State::Idle(mut connection) => {
                     match connection.poll_next_inbound(cx) {
                         Poll::Ready(maybe_stream) => {
-                            self.state = State::Idle(connection);
+                            // Transport layers will close the connection on the first
+                            // substream error. The `connection.poll_next_inbound` should
+                            // not be called again after returning an error. Instead, we
+                            // must close the connection gracefully.
+                            match maybe_stream.as_ref() {
+                                Some(Err(error)) => {
+                                    tracing::debug!(target: LOG_TARGET, ?error, "Inbound stream error, closing connection");
+
+                                    self.state = State::Closing {
+                                        reply: None,
+                                        inner: Closing::DrainingControlCommands { connection },
+                                    };
+                                }
+                                other => {
+                                    tracing::debug!(target: LOG_TARGET, ?other, "Inbound stream reset state to idle");
+                                    self.state = State::Idle(connection)
+                                }
+                            }
+
                             return Poll::Ready(maybe_stream);
                         }
                         Poll::Pending => {}
@@ -191,7 +211,7 @@ where
                         return Poll::Pending;
                     }
                 },
-                State::Poisoned => unreachable!(),
+                State::Poisoned => return Poll::Pending,
             }
         }
     }