flatten error

Author: rklaehn

iroh-connection-pool/src/connection_pool.rs

@@ -9,7 +9,10 @@
 //! It is important that you use the connection only in the future passed to
 //! connect, and don't clone it out of the future.
 use std::{
-    collections::{HashMap, VecDeque}, ops::Deref, sync::Arc, time::Duration
+    collections::{HashMap, VecDeque},
+    ops::Deref,
+    sync::Arc,
+    time::Duration,
 };
 
 use iroh::{
@@ -19,7 +22,11 @@ use iroh::{
 use n0_future::MaybeFuture;
 use snafu::Snafu;
 use tokio::{
-    sync::{mpsc::{self, error::SendError as TokioSendError}, oneshot, OwnedSemaphorePermit},
+    sync::{
+        OwnedSemaphorePermit,
+        mpsc::{self, error::SendError as TokioSendError},
+        oneshot,
+    },
     task::JoinError,
 };
 use tokio_util::time::FutureExt;
@@ -80,25 +87,25 @@ struct Context {
 /// errors such as timeouts and connection limits.
 #[derive(Debug, Clone)]
 pub enum PoolConnectError {
+    /// Connection pool is shut down
+    Shutdown,
     /// Timeout during connect
     Timeout,
     /// Too many connections
     TooManyConnections,
     /// Error during connect
     ConnectError(Arc<ConnectError>),
-    /// Error during last execute
-    ExecuteError(Arc<ExecuteError>),
     /// Handler actor panicked
     JoinError(Arc<JoinError>),
 }
 
 impl std::fmt::Display for PoolConnectError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
+            PoolConnectError::Shutdown => write!(f, "Connection pool is shut down"),
             PoolConnectError::Timeout => write!(f, "Connection timed out"),
             PoolConnectError::TooManyConnections => write!(f, "Too many connections"),
             PoolConnectError::ConnectError(e) => write!(f, "Connection error: {}", e),
-            PoolConnectError::ExecuteError(e) => write!(f, "Execution error: {}", e),
             PoolConnectError::JoinError(e) => write!(f, "Join error: {}", e),
         }
     }
@@ -286,8 +293,7 @@ impl Actor {
                             trace!("removing oldest idle connection {}", idle);
                             self.connections.remove(&idle);
                         } else {
-                            msg.tx.send(Err(PoolConnectError::TooManyConnections))
-                                .ok();
+                            msg.tx.send(Err(PoolConnectError::TooManyConnections)).ok();
                             continue;
                         }
                     }
@@ -327,19 +333,6 @@ pub enum ConnectionPoolError {
     Shutdown,
 }
 
-/// An error during the usage of the connection.
-///
-/// The connection pool will recreate the connection if a handler returns this
-/// error. If you don't want this, swallow the error in the handler.
-#[derive(Debug, Snafu)]
-pub struct ExecuteError;
-
-impl From<PoolConnectError> for ExecuteError {
-    fn from(_: PoolConnectError) -> Self {
-        ExecuteError
-    }
-}
-
 /// A connection pool
 #[derive(Debug, Clone)]
 pub struct ConnectionPool {
@@ -359,14 +352,13 @@ impl ConnectionPool {
     pub async fn connect(
         &self,
         id: NodeId,
-    ) -> std::result::Result<std::result::Result<ConnectionRef, PoolConnectError>, ConnectionPoolError>
-    {
+    ) -> std::result::Result<ConnectionRef, PoolConnectError> {
         let (tx, rx) = oneshot::channel();
         self.tx
             .send(ActorMessage::RequestRef(RequestRef { id, tx }))
             .await
-            .map_err(|_| ConnectionPoolError::Shutdown)?;
-        Ok(rx.await.map_err(|_| ConnectionPoolError::Shutdown)?)
+            .map_err(|_| PoolConnectError::Shutdown)?;
+        Ok(rx.await.map_err(|_| PoolConnectError::Shutdown)??)
     }
 
     /// Close an existing connection, if it exists

iroh-connection-pool/src/tests.rs

@@ -11,7 +11,7 @@ use n0_snafu::ResultExt;
 use testresult::TestResult;
 use tracing::trace;
 
-use crate::connection_pool::{ConnectionPool, ConnectionPoolError, Options, PoolConnectError};
+use crate::connection_pool::{ConnectionPool, Options, PoolConnectError};
 
 const ECHO_ALPN: &[u8] = b"echo";
 
@@ -92,27 +92,23 @@ impl EchoClient {
         &self,
         id: NodeId,
         text: Vec<u8>,
-    ) -> Result<
-        Result<Result<(usize, Vec<u8>), n0_snafu::Error>, PoolConnectError>,
-        ConnectionPoolError,
-    > {
+    ) -> Result<Result<(usize, Vec<u8>), n0_snafu::Error>, PoolConnectError> {
         let conn = self.pool.connect(id).await?;
-        let conn = match conn {
-            Ok(conn) => conn,
-            Err(e) => return Ok(Err(e)),
-        };
         let id = conn.stable_id();
         match echo_client(&conn, &text).await {
-            Ok(res) => Ok(Ok(Ok((id, res)))),
-            Err(e) => Ok(Ok(Err(e))),
+            Ok(res) => Ok(Ok((id, res))),
+            Err(e) => Ok(Err(e)),
         }
     }
 }
 
 #[tokio::test]
 async fn connection_pool_errors() -> TestResult<()> {
     let filter = tracing_subscriber::EnvFilter::from_default_env();
-    tracing_subscriber::fmt().with_env_filter(filter).try_init().ok();
+    tracing_subscriber::fmt()
+        .with_env_filter(filter)
+        .try_init()
+        .ok();
     // set up static discovery for all addrs
     let discovery = StaticProvider::new();
     let endpoint = iroh::Endpoint::builder()
@@ -123,7 +119,7 @@ async fn connection_pool_errors() -> TestResult<()> {
     let client = EchoClient { pool };
     {
         let non_existing = SecretKey::from_bytes(&[0; 32]).public();
-        let res = client.echo(non_existing, b"Hello, world!".to_vec()).await?;
+        let res = client.echo(non_existing, b"Hello, world!".to_vec()).await;
         // trying to connect to a non-existing id will fail with ConnectError
         // because we don't have any information about the node
         assert!(matches!(res, Err(PoolConnectError::ConnectError(_))));
@@ -140,9 +136,7 @@ async fn connection_pool_errors() -> TestResult<()> {
         });
         // trying to connect to an id for which we have info, but the other
         // end is not listening, will lead to a timeout.
-        let res = client
-            .echo(non_listening, b"Hello, world!".to_vec())
-            .await?;
+        let res = client.echo(non_listening, b"Hello, world!".to_vec()).await;
         assert!(matches!(res, Err(PoolConnectError::Timeout)));
     }
     Ok(())
@@ -151,7 +145,10 @@ async fn connection_pool_errors() -> TestResult<()> {
 #[tokio::test]
 async fn connection_pool_smoke() -> TestResult<()> {
     let filter = tracing_subscriber::EnvFilter::from_default_env();
-    tracing_subscriber::fmt().with_env_filter(filter).try_init().ok();
+    tracing_subscriber::fmt()
+        .with_env_filter(filter)
+        .try_init()
+        .ok();
     let n = 32;
     let nodes = echo_servers(n).await?;
     let ids = nodes
@@ -170,17 +167,17 @@ async fn connection_pool_smoke() -> TestResult<()> {
     let mut connection_ids = BTreeMap::new();
     let msg = b"Hello, world!".to_vec();
     for id in &ids {
-        let (cid1, res) = client.echo(*id, msg.clone()).await???;
+        let (cid1, res) = client.echo(*id, msg.clone()).await??;
         assert_eq!(res, msg);
-        let (cid2, res) = client.echo(*id, msg.clone()).await???;
+        let (cid2, res) = client.echo(*id, msg.clone()).await??;
         assert_eq!(res, msg);
         assert_eq!(cid1, cid2);
         connection_ids.insert(id, cid1);
     }
     tokio::time::sleep(Duration::from_millis(1000)).await;
     for id in &ids {
         let cid1 = *connection_ids.get(id).expect("Connection ID not found");
-        let (cid2, res) = client.echo(*id, msg.clone()).await???;
+        let (cid2, res) = client.echo(*id, msg.clone()).await??;
         assert_eq!(res, msg);
         assert_ne!(cid1, cid2);
     }
@@ -192,7 +189,10 @@ async fn connection_pool_smoke() -> TestResult<()> {
 #[tokio::test]
 async fn connection_pool_idle() -> TestResult<()> {
     let filter = tracing_subscriber::EnvFilter::from_default_env();
-    tracing_subscriber::fmt().with_env_filter(filter).try_init().ok();
+    tracing_subscriber::fmt()
+        .with_env_filter(filter)
+        .try_init()
+        .ok();
     let n = 32;
     let nodes = echo_servers(n).await?;
     let ids = nodes
@@ -218,7 +218,7 @@ async fn connection_pool_idle() -> TestResult<()> {
     let client = EchoClient { pool };
     let msg = b"Hello, world!".to_vec();
     for id in &ids {
-        let (_, res) = client.echo(*id, msg.clone()).await???;
+        let (_, res) = client.echo(*id, msg.clone()).await??;
         assert_eq!(res, msg);
     }
     Ok(())