transports/tcp/: Call take_error on TCP socket (libp2p#2458)

Within `Provider::new_stream` we wait for the socket to become writable
(`stream.writable`), before returning it as a stream. In other words, we
are waiting for the socket to connect before returning it as a new TCP
connection.  Waiting to connect before returning it as a new TCP
connection allows us to catch TCP connection establishment errors early.

While `stream.writable` drives the process of connecting, it does not
surface potential connection errors themselves. These need to be
explicitly collected via `TcpSocket::take_error`. If not explicitly
collected, they will surface on future operations on the socket.

For now this commit explicitly calls `TcpSocket::take_error` when using
`async-io` only. `tokio` introduced the method (`take_error`) in
tokio-rs/tokio#4364 though later reverted it in
tokio-rs/tokio#4392. Once re-reverted, the same
patch can be applied when using `libp2p-tcp` with tokio.

---

One example on how this bug surfaces today:

A `/dnsaddr/xxx` `Multiaddr` can potentially resolve to multiple IP
addresses, e.g. to the IPv4 and the IPv6 addresses of a node.
`libp2p-dns` tries dialing each of them in sequence using `libp2p-tcp`,
returning the first that `libp2p-tcp` reports as successful.

Say that the local node tries the IPv6 address first. In the scenario
where the local node's networking stack does not support IPv6, e.g. has
no IPv6 route, the connection attempt to the resolved IPv6 address of
the remote node fails. Given that `libp2p-tcp` does not call
`TcpSocket::take_error`, it would falsly report the TCP connection
attempt as successful. `libp2p-dns` would receive the "successful" TCP
connection for the IPv6 address from `libp2p-tcp` and would not attempt
to dial the IPv4 address, even though it supports IPv4, and instead
bubble up the "successful" IPv6 TCP connection. Only later, when writing
or reading from the "successful" IPv6 TCP connection, would the IPv6
error surface.

Co-authored-by: Oliver Wangler <oliver@wngr.de>

CHANGELOG.md

@@ -46,6 +46,7 @@
 
 - Update individual crates.
   - `libp2p-relay`
+  - `libp2p-tcp`
 
 ## Version 0.42.0 [2022-01-27]
 

Cargo.toml

@@ -106,7 +106,7 @@ smallvec = "1.6.1"
 libp2p-deflate = { version = "0.31.0", path = "transports/deflate", optional = true }
 libp2p-dns = { version = "0.31.0", path = "transports/dns", optional = true, default-features = false }
 libp2p-mdns = { version = "0.34.0", path = "protocols/mdns", optional = true }
-libp2p-tcp = { version = "0.31.0", path = "transports/tcp", default-features = false, optional = true }
+libp2p-tcp = { version = "0.31.1", path = "transports/tcp", default-features = false, optional = true }
 libp2p-websocket = { version = "0.33.0", path = "transports/websocket", optional = true }
 
 [dev-dependencies]

transports/tcp/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.31.1 [unreleased]
+
+- Call `TcpSocket::take_error` to report connection establishment errors early.
+
 # 0.31.0 [2022-01-27]
 
 - Update dependencies.

transports/tcp/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-tcp"
 edition = "2021"
 rust-version = "1.56.1"
 description = "TCP/IP transport protocol for libp2p"
-version = "0.31.0"
+version = "0.31.1"
 authors = ["Parity Technologies <admin@parity.io>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

transports/tcp/src/lib.rs

@@ -852,6 +852,8 @@ mod tests {
                             panic!("No TCP port in address: {}", a)
                         }
                         ready_tx.send(a).await.ok();
+                    }
+                    ListenerEvent::Upgrade { .. } => {
                         return;
                     }
                     _ => {}

transports/tcp/src/provider/async_io.rs

@@ -44,9 +44,18 @@ impl Provider for Tcp {
 
     fn new_stream(s: net::TcpStream) -> BoxFuture<'static, io::Result<Self::Stream>> {
         async move {
+            // Taken from [`Async::connect`].
+
             let stream = Async::new(s)?;
+
+            // The stream becomes writable when connected.
             stream.writable().await?;
-            Ok(stream)
+
+            // Check if there was an error while connecting.
+            match stream.get_ref().take_error()? {
+                None => Ok(stream),
+                Some(err) => Err(err),
+            }
         }
         .boxed()
     }