fmt

Author: rklaehn

iroh-connection-pool/src/connection_pool.rs

@@ -91,10 +91,9 @@ async fn run_connection_actor(
         Ok(Err(e)) => Err(PoolConnectError::ConnectError(e)),
         Err(_) => Err(PoolConnectError::Timeout),
     };
-    if state.is_err()
-        && context.owner.close(node_id).await.is_err() {
-            return;
-        }
+    if state.is_err() && context.owner.close(node_id).await.is_err() {
+        return;
+    }
 
     let mut tasks = JoinSet::new();
     let idle_timer = MaybeFuture::default();
@@ -120,30 +119,27 @@ async fn run_connection_actor(
             }
 
             // Handle completed tasks
-            task_result = tasks.join_next(), if !tasks.is_empty() => {
+            Some(task_result) = tasks.join_next(), if !tasks.is_empty() => {
                 match task_result {
-                    Some(Ok(Ok(()))) => {
+                    (Ok(Ok(()))) => {
                         tracing::debug!("Task completed for node {}", node_id);
                     }
-                    Some(Ok(Err(e))) => {
+                    (Ok(Err(e))) => {
                         tracing::error!("Task failed for node {}: {}", node_id, e);
                         if let Ok(conn) = state {
-                            conn.close(1u32.into(), b"");
+                            conn.close(1u32.into(), b"error");
                         }
                         state = Err(PoolConnectError::ExecuteError(e));
                         let _ = context.owner.close(node_id).await;
                     }
-                    Some(Err(e)) => {
+                    (Err(e)) => {
                         tracing::error!("Task panicked for node {}: {}", node_id, e);
                         if let Ok(conn) = state {
-                            conn.close(1u32.into(), b"");
+                            conn.close(1u32.into(), b"panic");
                         }
                         state = Err(PoolConnectError::JoinError(e));
                         let _ = context.owner.close(node_id).await;
                     }
-                    None => {
-                        tracing::debug!("Task was cancelled or already completed for node {}", node_id);
-                    }
                 }
 
                 // We are idle
@@ -175,7 +171,7 @@ async fn run_connection_actor(
     }
 
     if let Ok(connection) = &state {
-        connection.close(0u32.into(), b"");
+        connection.close(0u32.into(), b"idle");
     }
 
     tracing::debug!("Connection actor for {} shutting down", node_id);

iroh-connection-pool/src/connection_pool_0rtt.rs

@@ -141,10 +141,9 @@ async fn run_connection_actor(
         Ok((state, forwarder)) => (state, forwarder),
         Err(_) => (Err(PoolConnectError::Timeout), MaybeFuture::None),
     };
-    if state.is_err()
-        && context.owner.close(node_id).await.is_err() {
-            return;
-        }
+    if state.is_err() && context.owner.close(node_id).await.is_err() {
+        return;
+    }
     let mut tasks = JoinSet::new();
     let idle_timer = MaybeFuture::default();
     tokio::pin!(idle_timer);
@@ -169,30 +168,27 @@ async fn run_connection_actor(
             }
 
             // Handle completed tasks
-            task_result = tasks.join_next(), if !tasks.is_empty() => {
+            Some(task_result) = tasks.join_next(), if !tasks.is_empty() => {
                 match task_result {
-                    Some(Ok(Ok(()))) => {
+                    Ok(Ok(())) => {
                         trace!("Task completed for node {}", node_id);
                     }
-                    Some(Ok(Err(e))) => {
+                    Ok(Err(e)) => {
                         trace!("Task failed for node {}: {}", node_id, e);
                         if let Ok((conn, _)) = state {
-                            conn.close(1u32.into(), b"");
+                            conn.close(1u32.into(), b"error");
                         }
                         state = Err(PoolConnectError::ExecuteError(e));
                         context.owner.close(node_id).await.ok();
                     }
-                    Some(Err(e)) => {
+                    Err(e) => {
                         error!("Task panicked for node {}: {}", node_id, e);
                         if let Ok((conn, _)) = state {
-                            conn.close(1u32.into(), b"");
+                            conn.close(1u32.into(), b"panic");
                         }
                         state = Err(PoolConnectError::JoinError(e));
                         context.owner.close(node_id).await.ok();
                     }
-                    None => {
-                        trace!("Task was cancelled or already completed for node {}", node_id);
-                    }
                 }
 
                 // We are idle
@@ -226,7 +222,7 @@ async fn run_connection_actor(
     }
 
     if let Ok((conn, _)) = state {
-        conn.close(0u32.into(), b"");
+        conn.close(0u32.into(), b"idle");
     }
 
     trace!("Connection actor for {} shutting down", node_id);