Merge #238

238: Bob error handling r=thomaseizinger a=da-kami



Co-authored-by: Daniel Karzel <daniel@comit.network>

swap/src/bin/swap_cli.rs

@@ -128,8 +128,16 @@ async fn main() -> Result<()> {
             );
             let (swap, event_loop) = bob_factory.with_init_params(send_bitcoin).build().await?;
 
-            tokio::spawn(async move { event_loop.run().await });
-            bob::run(swap).await?;
+            let handle = tokio::spawn(async move { event_loop.run().await });
+            let swap = bob::run(swap);
+            tokio::select! {
+                event_loop_result = handle => {
+                    event_loop_result??;
+                },
+                swap_result = swap => {
+                    swap_result?;
+                }
+            }
         }
         Command::History => {
             let mut table = Table::new();
@@ -169,9 +177,16 @@ async fn main() -> Result<()> {
                 execution_params,
             );
             let (swap, event_loop) = bob_factory.build().await?;
-
-            tokio::spawn(async move { event_loop.run().await });
-            bob::run(swap).await?;
+            let handle = tokio::spawn(async move { event_loop.run().await });
+            let swap = bob::run(swap);
+            tokio::select! {
+                event_loop_result = handle => {
+                    event_loop_result??;
+                },
+                swap_result = swap => {
+                    swap_result?;
+                }
+            }
         }
         Command::Cancel {
             swap_id,
@@ -201,18 +216,22 @@ async fn main() -> Result<()> {
                 execution_params,
             );
             let (swap, event_loop) = bob_factory.build().await?;
+            let handle = tokio::spawn(async move { event_loop.run().await });
 
-            tokio::spawn(async move { event_loop.run().await });
-
-            match bob::cancel(
+            let cancel = bob::cancel(
                 swap.swap_id,
                 swap.state,
                 swap.bitcoin_wallet,
                 swap.db,
                 force,
-            )
-            .await?
-            {
+            );
+
+            tokio::select! {
+                event_loop_result = handle => {
+                    event_loop_result??;
+                },
+                cancel_result = cancel => {
+                    match cancel_result? {
                         Ok((txid, _)) => {
                             info!("Cancel transaction successfully published with id {}", txid)
                         }
@@ -225,6 +244,8 @@ async fn main() -> Result<()> {
                         }
                     }
                 }
+            }
+        }
         Command::Refund {
             swap_id,
             alice_peer_id,
@@ -254,19 +275,26 @@ async fn main() -> Result<()> {
             );
             let (swap, event_loop) = bob_factory.build().await?;
 
-            tokio::spawn(async move { event_loop.run().await });
-            bob::refund(
+            let handle = tokio::spawn(async move { event_loop.run().await });
+            let refund = bob::refund(
                 swap.swap_id,
                 swap.state,
                 swap.execution_params,
                 swap.bitcoin_wallet,
                 swap.db,
                 force,
-            )
-            .await??;
+            );
+
+            tokio::select! {
+                event_loop_result = handle => {
+                    event_loop_result??;
+                },
+                refund_result = refund => {
+                    refund_result??;
+                }
+            }
         }
     };
-
     Ok(())
 }
 

swap/src/protocol/bob.rs

@@ -185,7 +185,7 @@ pub enum OutEvent {
     },
     EncryptedSignatureAcknowledged,
     ResponseSent, // Same variant is used for all messages as no processing is done
-    Failure(Error),
+    CommunicationError(Error),
 }
 
 impl From<peer_tracker::OutEvent> for OutEvent {
@@ -203,7 +203,7 @@ impl From<quote_request::OutEvent> for OutEvent {
         use quote_request::OutEvent::*;
         match event {
             MsgReceived(quote_response) => OutEvent::QuoteResponse(quote_response),
-            Failure(err) => OutEvent::Failure(err.context("Failure with Quote Request")),
+            Failure(err) => OutEvent::CommunicationError(err.context("Failure with Quote Request")),
         }
     }
 }
@@ -225,7 +225,9 @@ impl From<transfer_proof::OutEvent> for OutEvent {
                 channel,
             },
             AckSent => OutEvent::ResponseSent,
-            Failure(err) => OutEvent::Failure(err.context("Failure with Transfer Proof")),
+            Failure(err) => {
+                OutEvent::CommunicationError(err.context("Failure with Transfer Proof"))
+            }
         }
     }
 }
@@ -235,7 +237,9 @@ impl From<encrypted_signature::OutEvent> for OutEvent {
         use encrypted_signature::OutEvent::*;
         match event {
             Acknowledged => OutEvent::EncryptedSignatureAcknowledged,
-            Failure(err) => OutEvent::Failure(err.context("Failure with Encrypted Signature")),
+            Failure(err) => {
+                OutEvent::CommunicationError(err.context("Failure with Encrypted Signature"))
+            }
         }
     }
 }

swap/src/protocol/bob/event_loop.rs

@@ -7,10 +7,10 @@ use crate::{
         bob::{Behaviour, OutEvent, QuoteRequest, State0, State2},
     },
 };
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, bail, Result};
 use futures::FutureExt;
 use libp2p::{core::Multiaddr, PeerId};
-use std::sync::Arc;
+use std::{convert::Infallible, sync::Arc};
 use tokio::sync::mpsc::{Receiver, Sender};
 use tracing::{debug, error, info};
 
@@ -167,7 +167,7 @@ impl EventLoop {
         Ok((event_loop, handle))
     }
 
-    pub async fn run(mut self) {
+    pub async fn run(mut self) -> Result<Infallible> {
         loop {
             tokio::select! {
                 swarm_event = self.swarm.next().fuse() => {
@@ -192,8 +192,8 @@ impl EventLoop {
                             debug!("Alice acknowledged encrypted signature");
                         }
                         OutEvent::ResponseSent => {}
-                        OutEvent::Failure(err) => {
-                            error!("Communication error: {:#}", err)
+                        OutEvent::CommunicationError(err) => {
+                            bail!("Communication error: {:#}", err)
                         }
                     }
                 },

swap/tests/testutils/mod.rs

@@ -9,6 +9,7 @@ use get_port::get_port;
 use libp2p::{core::Multiaddr, PeerId};
 use monero_harness::{image, Monero};
 use std::{
+    convert::Infallible,
     path::{Path, PathBuf},
     sync::Arc,
     time::Duration,
@@ -67,7 +68,7 @@ impl BobParams {
     }
 }
 
-pub struct BobEventLoopJoinHandle(JoinHandle<()>);
+pub struct BobEventLoopJoinHandle(JoinHandle<Result<Infallible>>);
 
 impl BobEventLoopJoinHandle {
     pub fn abort(&self) {