net/transport: Use io::Result in the net::transport module instead of crate::Result

src/net/transport/mod.rs

@@ -16,9 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use log::warn;
 use std::time::Duration;
-use tor_error::ErrorReport;
 
 use async_trait::async_trait;
 use smol::io::{AsyncRead, AsyncWrite};
@@ -221,18 +219,7 @@ impl Dialer {
             DialerVariant::Tor(dialer) => {
                 let host = self.endpoint.host_str().unwrap();
                 let port = self.endpoint.port().unwrap();
-                // Extract error reports (i.e. very detailed debugging)
-                // from arti-client in order to help debug Tor connections.
-                // https://docs.rs/arti-client/latest/arti_client/#reporting-arti-errors
-                // https://gitlab.torproject.org/tpo/core/arti/-/issues/1086
-                let result = match dialer.do_dial(host, port, timeout).await {
-                    Ok(stream) => Ok(stream),
-                    Err(err) => {
-                        warn!("{}", err.report());
-                        Err(err)
-                    }
-                };
-                let stream = result?;
+                let stream = dialer.do_dial(host, port, timeout).await?;
                 Ok(Box::new(stream))
             }
 
@@ -240,18 +227,7 @@ impl Dialer {
             DialerVariant::TorTls(dialer) => {
                 let host = self.endpoint.host_str().unwrap();
                 let port = self.endpoint.port().unwrap();
-                // Extract error reports (i.e. very detailed debugging)
-                // from arti-client in order to help debug Tor connections.
-                // https://docs.rs/arti-client/latest/arti_client/#reporting-arti-errors
-                // https://gitlab.torproject.org/tpo/core/arti/-/issues/1086
-                let result = match dialer.do_dial(host, port, timeout).await {
-                    Ok(stream) => Ok(stream),
-                    Err(err) => {
-                        warn!("{}", err.report());
-                        Err(err)
-                    }
-                };
-                let stream = result?;
+                let stream = dialer.do_dial(host, port, timeout).await?;
                 let tlsupgrade = tls::TlsUpgrade::new().await;
                 let stream = tlsupgrade.upgrade_dialer_tls(stream).await?;
                 Ok(Box::new(stream))

src/net/transport/tcp.rs

@@ -33,7 +33,6 @@ use socket2::{Domain, Socket, TcpKeepalive, Type};
 use url::Url;
 
 use super::{PtListener, PtStream};
-use crate::{Error, Result};
 
 /// TCP Dialer implementation
 #[derive(Debug, Clone)]
@@ -44,7 +43,7 @@ pub struct TcpDialer {
 
 impl TcpDialer {
     /// Instantiate a new [`TcpDialer`] with optional TTL.
-    pub(crate) async fn new(ttl: Option<u32>) -> Result<Self> {
+    pub(crate) async fn new(ttl: Option<u32>) -> io::Result<Self> {
         Ok(Self { ttl })
     }
 
@@ -74,7 +73,7 @@ impl TcpDialer {
         &self,
         socket_addr: SocketAddr,
         timeout: Option<Duration>,
-    ) -> Result<TcpStream> {
+    ) -> io::Result<TcpStream> {
         debug!(target: "net::tcp::do_dial", "Dialing {} with TCP...", socket_addr);
         let socket = self.create_socket(socket_addr).await?;
 
@@ -86,7 +85,7 @@ impl TcpDialer {
             Ok(()) => {}
             Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) => {}
             Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
-            Err(err) => return Err(err.into()),
+            Err(err) => return Err(err),
         };
 
         // Wrap socket in an Async wrapper.
@@ -95,7 +94,7 @@ impl TcpDialer {
         // Wait until the async object becomes writable.
         let connect = async {
             match async_socket.get_ref().take_error()? {
-                Some(err) => Err(Error::Io(err.kind())),
+                Some(err) => Err(err),
                 None => Ok(()),
             }
         };
@@ -122,7 +121,7 @@ impl TcpDialer {
                     }
                     Either::Left((Err(e), _)) => Err(e),
 
-                    Either::Right((_, _)) => Err(Error::ConnectTimeout),
+                    Either::Right((_, _)) => Err(io::ErrorKind::TimedOut.into()),
                 }
             }
             None => {
@@ -149,7 +148,7 @@ pub struct TcpListener {
 
 impl TcpListener {
     /// Instantiate a new [`TcpListener`] with given backlog size.
-    pub async fn new(backlog: i32) -> Result<Self> {
+    pub async fn new(backlog: i32) -> io::Result<Self> {
         Ok(Self { backlog })
     }
 
@@ -171,7 +170,7 @@ impl TcpListener {
     }
 
     /// Internal listen function
-    pub(crate) async fn do_listen(&self, socket_addr: SocketAddr) -> Result<SmolTcpListener> {
+    pub(crate) async fn do_listen(&self, socket_addr: SocketAddr) -> io::Result<SmolTcpListener> {
         let socket = self.create_socket(socket_addr).await?;
         socket.bind(&socket_addr.into())?;
         socket.listen(self.backlog)?;
@@ -186,7 +185,7 @@ impl TcpListener {
 
 #[async_trait]
 impl PtListener for SmolTcpListener {
-    async fn next(&self) -> std::io::Result<(Box<dyn PtStream>, Url)> {
+    async fn next(&self) -> io::Result<(Box<dyn PtStream>, Url)> {
         let (stream, peer_addr) = match self.accept().await {
             Ok((s, a)) => (s, a),
             Err(e) => return Err(e),
@@ -199,7 +198,7 @@ impl PtListener for SmolTcpListener {
 
 #[async_trait]
 impl PtListener for (TlsAcceptor, SmolTcpListener) {
-    async fn next(&self) -> std::io::Result<(Box<dyn PtStream>, Url)> {
+    async fn next(&self) -> io::Result<(Box<dyn PtStream>, Url)> {
         let (stream, peer_addr) = match self.1.accept().await {
             Ok((s, a)) => (s, a),
             Err(e) => return Err(e),

src/net/transport/tls.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
+use std::{io, sync::Arc};
 
 use futures_rustls::{
     rustls::{
@@ -36,8 +36,6 @@ use x509_parser::{
     prelude::{GeneralName, ParsedExtension, X509Certificate},
 };
 
-use crate::Result;
-
 /// Validate certificate DNSName.
 fn validate_dnsname(cert: &X509Certificate) -> std::result::Result<(), rustls::Error> {
     #[rustfmt::skip]
@@ -296,7 +294,7 @@ impl TlsUpgrade {
         Self { server_config, client_config }
     }
 
-    pub async fn upgrade_dialer_tls<IO>(self, stream: IO) -> Result<TlsStream<IO>>
+    pub async fn upgrade_dialer_tls<IO>(self, stream: IO) -> io::Result<TlsStream<IO>>
     where
         IO: super::PtStream,
     {
@@ -312,7 +310,7 @@ impl TlsUpgrade {
     pub async fn upgrade_listener_tcp_tls(
         self,
         listener: smol::net::TcpListener,
-    ) -> Result<(TlsAcceptor, smol::net::TcpListener)> {
+    ) -> io::Result<(TlsAcceptor, smol::net::TcpListener)> {
         Ok((TlsAcceptor::from(self.server_config), listener))
     }
 }

src/net/transport/tor.rs

@@ -16,15 +16,21 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::time::Duration;
+use std::{
+    io::{self, ErrorKind},
+    time::Duration,
+};
 
 use arti_client::{config::BoolOrAuto, DataStream, StreamPrefs, TorClient};
-use log::debug;
-use smol::lock::OnceCell;
+use futures::{
+    future::{select, Either},
+    pin_mut,
+};
+use log::{debug, warn};
+use smol::{lock::OnceCell, Timer};
+use tor_error::ErrorReport;
 use tor_rtcompat::PreferredRuntime;
 
-use crate::{system::timeout::timeout, Result};
-
 /// A static for `TorClient` reusability
 static TOR_CLIENT: OnceCell<TorClient<PreferredRuntime>> = OnceCell::new();
 
@@ -34,7 +40,7 @@ pub struct TorDialer;
 
 impl TorDialer {
     /// Instantiate a new [`TorDialer`] object
-    pub(crate) async fn new() -> Result<Self> {
+    pub(crate) async fn new() -> io::Result<Self> {
         Ok(Self {})
     }
 
@@ -44,27 +50,72 @@ impl TorDialer {
         host: &str,
         port: u16,
         conn_timeout: Option<Duration>,
-    ) -> Result<DataStream> {
+    ) -> io::Result<DataStream> {
         debug!(target: "net::tor::do_dial", "Dialing {}:{} with Tor...", host, port);
 
         // Initialize or fetch the static TOR_CLIENT that should be reused in
         // the Tor dialer
-        let client = TOR_CLIENT
+        let client = match TOR_CLIENT
             .get_or_try_init(|| async {
                 debug!(target: "net::tor::do_dial", "Bootstrapping...");
                 TorClient::builder().create_bootstrapped().await
             })
-            .await?;
+            .await
+        {
+            Ok(client) => client,
+            Err(e) => {
+                warn!("{}", e.report());
+                return Err(io::Error::new(
+                    ErrorKind::Other,
+                    "Internal Tor error, see logged warning",
+                ))
+            }
+        };
 
         let mut stream_prefs = StreamPrefs::new();
         stream_prefs.connect_to_onion_services(BoolOrAuto::Explicit(true));
 
-        let stream = if let Some(conn_timeout) = conn_timeout {
-            timeout(conn_timeout, client.connect_with_prefs((host, port), &stream_prefs)).await?
-        } else {
-            Ok(client.connect_with_prefs((host, port), &stream_prefs).await?)
-        };
+        // If a timeout is configured, run both the connect and timeout futures
+        // and return whatever finishes first. Otherwise, wait on the connect future.
+        let connect = client.connect_with_prefs((host, port), &stream_prefs);
+
+        match conn_timeout {
+            Some(t) => {
+                let timeout = Timer::after(t);
+                pin_mut!(timeout);
+                pin_mut!(connect);
+
+                match select(connect, timeout).await {
+                    Either::Left((Ok(stream), _)) => Ok(stream),
+
+                    Either::Left((Err(e), _)) => {
+                        warn!("{}", e.report());
+                        Err(io::Error::new(
+                            ErrorKind::Other,
+                            "Internal Tor error, see logged warning",
+                        ))
+                    }
+
+                    Either::Right((_, _)) => Err(io::ErrorKind::TimedOut.into()),
+                }
+            }
 
-        Ok(stream?)
+            None => {
+                match connect.await {
+                    Ok(stream) => Ok(stream),
+                    Err(e) => {
+                        // Extract error reports (i.e. very detailed debugging)
+                        // from arti-client in order to help debug Tor connections.
+                        // https://docs.rs/arti-client/latest/arti_client/#reporting-arti-errors
+                        // https://gitlab.torproject.org/tpo/core/arti/-/issues/1086
+                        warn!("{}", e.report());
+                        Err(io::Error::new(
+                            ErrorKind::Other,
+                            "Internal Tor error, see logged warning",
+                        ))
+                    }
+                }
+            }
+        }
     }
 }

src/net/transport/unix.rs

@@ -16,7 +16,10 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::path::{Path, PathBuf};
+use std::{
+    io,
+    path::{Path, PathBuf},
+};
 
 use async_trait::async_trait;
 use log::debug;
@@ -27,7 +30,6 @@ use smol::{
 use url::Url;
 
 use super::{PtListener, PtStream};
-use crate::Result;
 
 /// Unix Dialer implementation
 #[derive(Debug, Clone)]
@@ -35,7 +37,7 @@ pub struct UnixDialer;
 
 impl UnixDialer {
     /// Instantiate a new [`UnixDialer`] object
-    pub(crate) async fn new() -> Result<Self> {
+    pub(crate) async fn new() -> io::Result<Self> {
         Ok(Self {})
     }
 
@@ -43,7 +45,7 @@ impl UnixDialer {
     pub(crate) async fn do_dial(
         &self,
         path: impl AsRef<Path> + core::fmt::Debug,
-    ) -> Result<UnixStream> {
+    ) -> io::Result<UnixStream> {
         debug!(target: "net::unix::do_dial", "Dialing {:?} Unix socket...", path);
         let stream = UnixStream::connect(path).await?;
         Ok(stream)
@@ -56,12 +58,12 @@ pub struct UnixListener;
 
 impl UnixListener {
     /// Instantiate a new [`UnixListener`] object
-    pub(crate) async fn new() -> Result<Self> {
+    pub(crate) async fn new() -> io::Result<Self> {
         Ok(Self {})
     }
 
     /// Internal listen function
-    pub(crate) async fn do_listen(&self, path: &PathBuf) -> Result<SmolUnixListener> {
+    pub(crate) async fn do_listen(&self, path: &PathBuf) -> io::Result<SmolUnixListener> {
         // This rm is a bit aggressive, but c'est la vie.
         let _ = fs::remove_file(path).await;
         let listener = SmolUnixListener::bind(path)?;
@@ -71,7 +73,7 @@ impl UnixListener {
 
 #[async_trait]
 impl PtListener for SmolUnixListener {
-    async fn next(&self) -> std::io::Result<(Box<dyn PtStream>, Url)> {
+    async fn next(&self) -> io::Result<(Box<dyn PtStream>, Url)> {
         let (stream, _peer_addr) = match self.accept().await {
             Ok((s, a)) => (s, a),
             Err(e) => return Err(e),