Handle client error

protocols/perf/src/client.rs

@@ -24,6 +24,8 @@ mod handler;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 pub use behaviour::{Behaviour, Event};
+use libp2p_swarm::StreamUpgradeError;
+use void::Void;
 
 static NEXT_RUN_ID: AtomicUsize = AtomicUsize::new(1);
 
@@ -37,3 +39,11 @@ impl RunId {
         Self(NEXT_RUN_ID.fetch_add(1, Ordering::SeqCst))
     }
 }
+
+#[derive(thiserror::Error, Debug)]
+pub enum RunError {
+    #[error(transparent)]
+    Upgrade(#[from] StreamUpgradeError<Void>),
+    #[error("Failed to execute perf run: {0}")]
+    Io(#[from] std::io::Error),
+}

protocols/perf/src/client/behaviour.rs

@@ -29,20 +29,19 @@ use libp2p_core::Multiaddr;
 use libp2p_identity::PeerId;
 use libp2p_swarm::{
     derive_prelude::ConnectionEstablished, ConnectionClosed, ConnectionId, FromSwarm,
-    NetworkBehaviour, NotifyHandler, PollParameters, StreamUpgradeError, THandlerInEvent,
+    NetworkBehaviour, NotifyHandler, PollParameters, THandlerInEvent,
     THandlerOutEvent, ToSwarm,
 };
-use void::Void;
 
 use crate::RunParams;
 use crate::{client::handler::Handler, RunUpdate};
 
-use super::RunId;
+use super::{RunId, RunError};
 
 #[derive(Debug)]
 pub struct Event {
     pub id: RunId,
-    pub result: Result<RunUpdate, StreamUpgradeError<Void>>,
+    pub result: Result<RunUpdate, RunError>,
 }
 
 #[derive(Default)]

protocols/perf/src/client/handler.rs

@@ -33,12 +33,11 @@ use libp2p_swarm::{
         ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
         ListenUpgradeError,
     },
-    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, StreamProtocol, StreamUpgradeError,
-    SubstreamProtocol,
+    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, StreamProtocol, SubstreamProtocol,
 };
 use void::Void;
 
-use super::RunId;
+use super::{RunError, RunId};
 use crate::{RunParams, RunUpdate};
 
 #[derive(Debug)]
@@ -50,7 +49,7 @@ pub struct Command {
 #[derive(Debug)]
 pub struct Event {
     pub(crate) id: RunId,
-    pub(crate) result: Result<RunUpdate, StreamUpgradeError<Void>>,
+    pub(crate) result: Result<RunUpdate, RunError>,
 }
 
 pub struct Handler {
@@ -150,7 +149,7 @@ impl ConnectionHandler for Handler {
                 self.queued_events
                     .push_back(ConnectionHandlerEvent::NotifyBehaviour(Event {
                         id,
-                        result: Err(error),
+                        result: Err(error.into()),
                     }));
             }
             ConnectionEvent::ListenUpgradeError(ListenUpgradeError { info: (), error }) => {
@@ -179,10 +178,10 @@ impl ConnectionHandler for Handler {
             return Poll::Ready(event);
         }
 
-        while let Poll::Ready(Some((id, result))) = self.outbound.poll_next_unpin(cx) {
+        if let Poll::Ready(Some((id, result))) = self.outbound.poll_next_unpin(cx) {
             return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(Event {
                 id,
-                result: result.map_err(|e| todo!("{e:?}")),
+                result: result.map_err(Into::into),
             }));
         }