Error feedback for the user upon communication errors

If communication with the other party fails the program should stop and the user should see the respective error.
Communication errors are handled in the event-loop. Upon a communication error the event loop is stopped.
Since the event loop is only stopped upon error the Result returned from the event loop is Infallible.

If one of the two futures, event loop and swap,  finishes (success/failure) the other future should be stopped as well.
We use tokio::selec! to stop either future if the other stops.
This commit is contained in:
Daniel Karzel 2021-02-26 16:11:14 +11:00
parent e66e84085b
commit bb1537d6f2
4 changed files with 65 additions and 32 deletions

View File

@ -118,8 +118,16 @@ async fn main() -> Result<()> {
); );
let (swap, event_loop) = bob_factory.with_init_params(send_bitcoin).build().await?; let (swap, event_loop) = bob_factory.with_init_params(send_bitcoin).build().await?;
tokio::spawn(async move { event_loop.run().await }); let handle = tokio::spawn(async move { event_loop.run().await });
bob::run(swap).await?; let swap = bob::run(swap);
tokio::select! {
event_loop_result = handle => {
event_loop_result??;
},
swap_result = swap => {
swap_result?;
}
}
} }
Command::History => { Command::History => {
let mut table = Table::new(); let mut table = Table::new();
@ -159,9 +167,16 @@ async fn main() -> Result<()> {
execution_params, execution_params,
); );
let (swap, event_loop) = bob_factory.build().await?; 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 }); let swap = bob::run(swap);
bob::run(swap).await?; tokio::select! {
event_loop_result = handle => {
event_loop_result??;
},
swap_result = swap => {
swap_result?;
}
}
} }
Command::Cancel(Cancel::BuyXmr { Command::Cancel(Cancel::BuyXmr {
swap_id, swap_id,
@ -191,18 +206,22 @@ async fn main() -> Result<()> {
execution_params, execution_params,
); );
let (swap, event_loop) = bob_factory.build().await?; 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 }); let cancel = bob::cancel(
match bob::cancel(
swap.swap_id, swap.swap_id,
swap.state, swap.state,
swap.bitcoin_wallet, swap.bitcoin_wallet,
swap.db, swap.db,
force, force,
) );
.await?
{ tokio::select! {
event_loop_result = handle => {
event_loop_result??;
},
cancel_result = cancel => {
match cancel_result? {
Ok((txid, _)) => { Ok((txid, _)) => {
info!("Cancel transaction successfully published with id {}", txid) info!("Cancel transaction successfully published with id {}", txid)
} }
@ -215,6 +234,8 @@ async fn main() -> Result<()> {
} }
} }
} }
}
}
Command::Refund(Refund::BuyXmr { Command::Refund(Refund::BuyXmr {
swap_id, swap_id,
alice_peer_id, alice_peer_id,
@ -244,19 +265,26 @@ async fn main() -> Result<()> {
); );
let (swap, event_loop) = bob_factory.build().await?; let (swap, event_loop) = bob_factory.build().await?;
tokio::spawn(async move { event_loop.run().await }); let handle = tokio::spawn(async move { event_loop.run().await });
bob::refund( let refund = bob::refund(
swap.swap_id, swap.swap_id,
swap.state, swap.state,
swap.execution_params, swap.execution_params,
swap.bitcoin_wallet, swap.bitcoin_wallet,
swap.db, swap.db,
force, force,
) );
.await??;
tokio::select! {
event_loop_result = handle => {
event_loop_result??;
},
refund_result = refund => {
refund_result??;
}
}
} }
}; };
Ok(()) Ok(())
} }

View File

@ -225,7 +225,9 @@ impl From<transfer_proof::OutEvent> for OutEvent {
channel, channel,
}, },
AckSent => OutEvent::ResponseSent, AckSent => OutEvent::ResponseSent,
Failure(err) => OutEvent::CommunicationError(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::*; use encrypted_signature::OutEvent::*;
match event { match event {
Acknowledged => OutEvent::EncryptedSignatureAcknowledged, Acknowledged => OutEvent::EncryptedSignatureAcknowledged,
Failure(err) => OutEvent::CommunicationError(err.context("Failure with Encrypted Signature")), Failure(err) => {
OutEvent::CommunicationError(err.context("Failure with Encrypted Signature"))
}
} }
} }
} }

View File

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

View File

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