Alice spawns swaps outside the event loop

Instead of spawning the swap inside the event loop we send the swap back
to the caller to be spawned. This means we no longer need the remote handle
that was only used in the tests.
This now properly logs the swap results in production.
It also gives us more control over Alice's swap in the tests.
This commit is contained in:
Daniel Karzel 2021-03-16 17:08:19 +11:00
parent 904312d1e9
commit ea05c306e0
8 changed files with 93 additions and 72 deletions

View File

@ -27,7 +27,7 @@ use swap::database::Database;
use swap::execution_params::{ExecutionParams, GetExecutionParams}; use swap::execution_params::{ExecutionParams, GetExecutionParams};
use swap::fs::default_config_path; use swap::fs::default_config_path;
use swap::monero::Amount; use swap::monero::Amount;
use swap::protocol::alice::EventLoop; use swap::protocol::alice::{run, EventLoop};
use swap::seed::Seed; use swap::seed::Seed;
use swap::trace::init_tracing; use swap::trace::init_tracing;
use swap::{bitcoin, execution_params, kraken, monero}; use swap::{bitcoin, execution_params, kraken, monero};
@ -95,7 +95,7 @@ async fn main() -> Result<()> {
let kraken_rate_updates = kraken::connect()?; let kraken_rate_updates = kraken::connect()?;
let (event_loop, _) = EventLoop::new( let (event_loop, mut swap_receiver) = EventLoop::new(
config.network.listen, config.network.listen,
seed, seed,
execution_params, execution_params,
@ -107,6 +107,22 @@ async fn main() -> Result<()> {
) )
.unwrap(); .unwrap();
tokio::spawn(async move {
while let Some(swap) = swap_receiver.recv().await {
tokio::spawn(async move {
let swap_id = swap.swap_id;
match run(swap).await {
Ok(state) => {
tracing::debug!(%swap_id, "Swap finished with state {}", state)
}
Err(e) => {
tracing::error!(%swap_id, "Swap failed with {:#}", e)
}
}
});
}
});
info!("Our peer id is {}", event_loop.peer_id()); info!("Our peer id is {}", event_loop.peer_id());
event_loop.run().await; event_loop.run().await;

View File

@ -4,20 +4,17 @@ use crate::execution_params::ExecutionParams;
use crate::monero::BalanceTooLow; use crate::monero::BalanceTooLow;
use crate::network::quote::BidQuote; use crate::network::quote::BidQuote;
use crate::network::{spot_price, transport, TokioExecutor}; use crate::network::{spot_price, transport, TokioExecutor};
use crate::protocol::alice;
use crate::protocol::alice::{AliceState, Behaviour, OutEvent, State3, Swap, TransferProof}; use crate::protocol::alice::{AliceState, Behaviour, OutEvent, State3, Swap, TransferProof};
use crate::protocol::bob::EncryptedSignature; use crate::protocol::bob::EncryptedSignature;
use crate::seed::Seed; use crate::seed::Seed;
use crate::{bitcoin, kraken, monero}; use crate::{bitcoin, kraken, monero};
use anyhow::{bail, Context, Result}; use anyhow::{bail, Context, Result};
use futures::future::RemoteHandle;
use libp2p::core::Multiaddr; use libp2p::core::Multiaddr;
use libp2p::futures::FutureExt; use libp2p::futures::FutureExt;
use libp2p::{PeerId, Swarm}; use libp2p::{PeerId, Swarm};
use rand::rngs::OsRng; use rand::rngs::OsRng;
use std::convert::Infallible; use std::convert::Infallible;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::mpsc::error::SendError;
use tokio::sync::{broadcast, mpsc}; use tokio::sync::{broadcast, mpsc};
use tracing::{debug, error, trace}; use tracing::{debug, error, trace};
use uuid::Uuid; use uuid::Uuid;
@ -39,7 +36,7 @@ pub struct EventLoop<RS> {
// Only used to produce new handles // Only used to produce new handles
send_transfer_proof_sender: mpsc::Sender<(PeerId, TransferProof)>, send_transfer_proof_sender: mpsc::Sender<(PeerId, TransferProof)>,
swap_handle_sender: mpsc::Sender<RemoteHandle<Result<AliceState>>>, swap_sender: mpsc::Sender<Swap>,
} }
#[derive(Debug)] #[derive(Debug)]
@ -62,7 +59,7 @@ where
db: Arc<Database>, db: Arc<Database>,
latest_rate: LR, latest_rate: LR,
max_buy: bitcoin::Amount, max_buy: bitcoin::Amount,
) -> Result<(Self, mpsc::Receiver<RemoteHandle<Result<AliceState>>>)> { ) -> Result<(Self, mpsc::Receiver<Swap>)> {
let identity = seed.derive_libp2p_identity(); let identity = seed.derive_libp2p_identity();
let behaviour = Behaviour::default(); let behaviour = Behaviour::default();
let transport = transport::build(&identity)?; let transport = transport::build(&identity)?;
@ -79,7 +76,7 @@ where
let recv_encrypted_signature = BroadcastChannels::default(); let recv_encrypted_signature = BroadcastChannels::default();
let send_transfer_proof = MpscChannels::default(); let send_transfer_proof = MpscChannels::default();
let swap_handle = MpscChannels::default(); let swap_channel = MpscChannels::default();
let event_loop = EventLoop { let event_loop = EventLoop {
swarm, swarm,
@ -92,10 +89,10 @@ where
recv_encrypted_signature: recv_encrypted_signature.sender, recv_encrypted_signature: recv_encrypted_signature.sender,
send_transfer_proof: send_transfer_proof.receiver, send_transfer_proof: send_transfer_proof.receiver,
send_transfer_proof_sender: send_transfer_proof.sender, send_transfer_proof_sender: send_transfer_proof.sender,
swap_handle_sender: swap_handle.sender, swap_sender: swap_channel.sender,
max_buy, max_buy,
}; };
Ok((event_loop, swap_handle.receiver)) Ok((event_loop, swap_channel.receiver))
} }
pub fn new_handle(&self) -> EventLoopHandle { pub fn new_handle(&self) -> EventLoopHandle {
@ -231,11 +228,7 @@ where
}) })
} }
async fn handle_execution_setup_done( async fn handle_execution_setup_done(&mut self, bob_peer_id: PeerId, state3: State3) {
&mut self,
bob_peer_id: PeerId,
state3: State3,
) -> Result<()> {
let swap_id = Uuid::new_v4(); let swap_id = Uuid::new_v4();
let handle = self.new_handle(); let handle = self.new_handle();
@ -254,18 +247,9 @@ where
swap_id, swap_id,
}; };
let (swap, swap_handle) = alice::run(swap).remote_handle(); if let Err(error) = self.swap_sender.send(swap).await {
tokio::spawn(swap); tracing::warn!(%swap_id, "Swap cannot be spawned: {}", error);
// For testing purposes the handle is currently sent via a channel so we can
// await it. If a remote handle is dropped, the future of the swap is
// also stopped. If we error upon sending the handle through the channel
// we have to call forget to detach the handle from the swap future.
if let Err(SendError(handle)) = self.swap_handle_sender.send(swap_handle).await {
handle.forget();
} }
Ok(())
} }
} }

View File

@ -1,17 +1,20 @@
pub mod testutils; pub mod testutils;
use swap::protocol::bob;
use swap::protocol::bob::BobState; use swap::protocol::bob::BobState;
use swap::protocol::{alice, bob};
use testutils::bob_run_until::is_btc_locked; use testutils::bob_run_until::is_btc_locked;
use testutils::FastCancelConfig; use testutils::FastCancelConfig;
#[tokio::test] #[tokio::test]
async fn given_bob_manually_refunds_after_btc_locked_bob_refunds() { async fn given_bob_manually_refunds_after_btc_locked_bob_refunds() {
testutils::setup_test(FastCancelConfig, |mut ctx| async move { testutils::setup_test(FastCancelConfig, |mut ctx| async move {
let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await; let (bob_swap, bob_join_handle) = ctx.bob_swap().await;
let bob_swap = tokio::spawn(bob::run_until(bob_swap, is_btc_locked));
let bob_state = bob::run_until(bob_swap, is_btc_locked).await.unwrap(); let alice_swap = ctx.alice_next_swap().await;
let _ = tokio::spawn(alice::run(alice_swap));
let bob_state = bob_swap.await??;
assert!(matches!(bob_state, BobState::BtcLocked { .. })); assert!(matches!(bob_state, BobState::BtcLocked { .. }));
let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
@ -20,8 +23,7 @@ async fn given_bob_manually_refunds_after_btc_locked_bob_refunds() {
if let BobState::BtcLocked(state3) = bob_swap.state.clone() { if let BobState::BtcLocked(state3) = bob_swap.state.clone() {
state3 state3
.wait_for_cancel_timelock_to_expire(bob_swap.bitcoin_wallet.as_ref()) .wait_for_cancel_timelock_to_expire(bob_swap.bitcoin_wallet.as_ref())
.await .await?;
.unwrap();
} else { } else {
panic!("Bob in unexpected state {}", bob_swap.state); panic!("Bob in unexpected state {}", bob_swap.state);
} }
@ -35,9 +37,7 @@ async fn given_bob_manually_refunds_after_btc_locked_bob_refunds() {
bob_swap.db, bob_swap.db,
false, false,
) )
.await .await??;
.unwrap()
.unwrap();
assert!(matches!(state, BobState::BtcCancelled { .. })); assert!(matches!(state, BobState::BtcCancelled { .. }));
let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
@ -53,11 +53,11 @@ async fn given_bob_manually_refunds_after_btc_locked_bob_refunds() {
bob_swap.db, bob_swap.db,
false, false,
) )
.await .await??;
.unwrap()
.unwrap();
ctx.assert_bob_refunded(bob_state).await; ctx.assert_bob_refunded(bob_state).await;
Ok(())
}) })
.await; .await
} }

View File

@ -1,17 +1,21 @@
pub mod testutils; pub mod testutils;
use bob::cancel::Error; use bob::cancel::Error;
use swap::protocol::bob;
use swap::protocol::bob::BobState; use swap::protocol::bob::BobState;
use swap::protocol::{alice, bob};
use testutils::bob_run_until::is_btc_locked; use testutils::bob_run_until::is_btc_locked;
use testutils::SlowCancelConfig; use testutils::SlowCancelConfig;
#[tokio::test] #[tokio::test]
async fn given_bob_manually_cancels_when_timelock_not_expired_errors() { async fn given_bob_manually_cancels_when_timelock_not_expired_errors() {
testutils::setup_test(SlowCancelConfig, |mut ctx| async move { testutils::setup_test(SlowCancelConfig, |mut ctx| async move {
let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await; let (bob_swap, bob_join_handle) = ctx.bob_swap().await;
let bob_swap = tokio::spawn(bob::run_until(bob_swap, is_btc_locked));
let bob_state = bob::run_until(bob_swap, is_btc_locked).await.unwrap(); let alice_swap = ctx.alice_next_swap().await;
let _ = tokio::spawn(alice::run(alice_swap));
let bob_state = bob_swap.await??;
assert!(matches!(bob_state, BobState::BtcLocked { .. })); assert!(matches!(bob_state, BobState::BtcLocked { .. }));
let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
@ -25,8 +29,7 @@ async fn given_bob_manually_cancels_when_timelock_not_expired_errors() {
bob_swap.db, bob_swap.db,
false, false,
) )
.await .await?
.unwrap()
.err() .err()
.unwrap(); .unwrap();
@ -44,13 +47,14 @@ async fn given_bob_manually_cancels_when_timelock_not_expired_errors() {
bob_swap.db, bob_swap.db,
false, false,
) )
.await .await?
.unwrap()
.err() .err()
.unwrap(); .unwrap();
let (bob_swap, _) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, _) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
assert!(matches!(bob_swap.state, BobState::BtcLocked { .. })); assert!(matches!(bob_swap.state, BobState::BtcLocked { .. }));
Ok(())
}) })
.await; .await;
} }

View File

@ -1,16 +1,20 @@
pub mod testutils; pub mod testutils;
use swap::protocol::bob;
use swap::protocol::bob::BobState; use swap::protocol::bob::BobState;
use swap::protocol::{alice, bob};
use testutils::bob_run_until::is_btc_locked; use testutils::bob_run_until::is_btc_locked;
use testutils::SlowCancelConfig; use testutils::SlowCancelConfig;
#[tokio::test] #[tokio::test]
async fn given_bob_manually_forces_cancel_when_timelock_not_expired_errors() { async fn given_bob_manually_forces_cancel_when_timelock_not_expired_errors() {
testutils::setup_test(SlowCancelConfig, |mut ctx| async move { testutils::setup_test(SlowCancelConfig, |mut ctx| async move {
let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await; let (bob_swap, bob_join_handle) = ctx.bob_swap().await;
let bob_swap = tokio::spawn(bob::run_until(bob_swap, is_btc_locked));
let bob_state = bob::run_until(bob_swap, is_btc_locked).await.unwrap(); let alice_swap = ctx.alice_next_swap().await;
let _ = tokio::spawn(alice::run(alice_swap));
let bob_state = bob_swap.await??;
assert!(matches!(bob_state, BobState::BtcLocked { .. })); assert!(matches!(bob_state, BobState::BtcLocked { .. }));
let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, bob_join_handle) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
@ -47,6 +51,8 @@ async fn given_bob_manually_forces_cancel_when_timelock_not_expired_errors() {
assert!(is_error); assert!(is_error);
let (bob_swap, _) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, _) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
assert!(matches!(bob_swap.state, BobState::BtcLocked { .. })); assert!(matches!(bob_swap.state, BobState::BtcLocked { .. }));
Ok(())
}) })
.await; .await;
} }

View File

@ -1,19 +1,26 @@
pub mod testutils; pub mod testutils;
use swap::protocol::bob; use swap::protocol::{alice, bob};
use testutils::SlowCancelConfig; use testutils::SlowCancelConfig;
use tokio::join;
/// Run the following tests with RUST_MIN_STACK=10000000 /// Run the following tests with RUST_MIN_STACK=10000000
#[tokio::test] #[tokio::test]
async fn happy_path() { async fn happy_path() {
testutils::setup_test(SlowCancelConfig, |mut ctx| async move { testutils::setup_test(SlowCancelConfig, |mut ctx| async move {
let (bob_swap, _) = ctx.new_swap_as_bob().await; let (bob_swap, _) = ctx.bob_swap().await;
let bob_swap = tokio::spawn(bob::run(bob_swap));
let bob_state = bob::run(bob_swap).await; let alice_swap = ctx.alice_next_swap().await;
let alice_swap = tokio::spawn(alice::run(alice_swap));
ctx.assert_alice_redeemed().await; let (bob_state, alice_state) = join!(bob_swap, alice_swap);
ctx.assert_bob_redeemed(bob_state.unwrap()).await;
ctx.assert_alice_redeemed(alice_state??).await;
ctx.assert_bob_redeemed(bob_state??).await;
Ok(())
}) })
.await; .await;
} }

View File

@ -1,27 +1,34 @@
pub mod testutils; pub mod testutils;
use swap::protocol::bob;
use swap::protocol::bob::BobState; use swap::protocol::bob::BobState;
use swap::protocol::{alice, bob};
use testutils::bob_run_until::is_xmr_locked; use testutils::bob_run_until::is_xmr_locked;
use testutils::SlowCancelConfig; use testutils::SlowCancelConfig;
#[tokio::test] #[tokio::test]
async fn given_bob_restarts_after_xmr_is_locked_resume_swap() { async fn given_bob_restarts_after_xmr_is_locked_resume_swap() {
testutils::setup_test(SlowCancelConfig, |mut ctx| async move { testutils::setup_test(SlowCancelConfig, |mut ctx| async move {
let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await; let (bob_swap, bob_join_handle) = ctx.bob_swap().await;
let bob_swap = tokio::spawn(bob::run_until(bob_swap, is_xmr_locked));
let bob_state = bob::run_until(bob_swap, is_xmr_locked).await.unwrap(); let alice_swap = ctx.alice_next_swap().await;
let alice_swap = tokio::spawn(alice::run(alice_swap));
let bob_state = bob_swap.await??;
assert!(matches!(bob_state, BobState::XmrLocked { .. })); assert!(matches!(bob_state, BobState::XmrLocked { .. }));
let (bob_swap, _) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await; let (bob_swap, _) = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
assert!(matches!(bob_swap.state, BobState::XmrLocked { .. })); assert!(matches!(bob_swap.state, BobState::XmrLocked { .. }));
let bob_state = bob::run(bob_swap).await.unwrap(); let bob_state = bob::run(bob_swap).await?;
ctx.assert_bob_redeemed(bob_state).await; ctx.assert_bob_redeemed(bob_state).await;
ctx.assert_alice_redeemed().await; let alice_state = alice_swap.await??;
ctx.assert_alice_redeemed(alice_state).await;
Ok(())
}) })
.await; .await;
} }

View File

@ -4,7 +4,6 @@ mod electrs;
use crate::testutils; use crate::testutils;
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use bitcoin_harness::{BitcoindRpcApi, Client}; use bitcoin_harness::{BitcoindRpcApi, Client};
use futures::future::RemoteHandle;
use futures::Future; use futures::Future;
use get_port::get_port; use get_port::get_port;
use libp2p::core::Multiaddr; use libp2p::core::Multiaddr;
@ -18,7 +17,7 @@ use swap::asb::FixedRate;
use swap::bitcoin::{CancelTimelock, PunishTimelock}; use swap::bitcoin::{CancelTimelock, PunishTimelock};
use swap::database::Database; use swap::database::Database;
use swap::execution_params::{ExecutionParams, GetExecutionParams}; use swap::execution_params::{ExecutionParams, GetExecutionParams};
use swap::protocol::alice::AliceState; use swap::protocol::alice::{AliceState, Swap};
use swap::protocol::bob::BobState; use swap::protocol::bob::BobState;
use swap::protocol::{alice, bob}; use swap::protocol::{alice, bob};
use swap::seed::Seed; use swap::seed::Seed;
@ -98,7 +97,7 @@ pub struct TestContext {
alice_starting_balances: StartingBalances, alice_starting_balances: StartingBalances,
alice_bitcoin_wallet: Arc<bitcoin::Wallet>, alice_bitcoin_wallet: Arc<bitcoin::Wallet>,
alice_monero_wallet: Arc<monero::Wallet>, alice_monero_wallet: Arc<monero::Wallet>,
alice_swap_handle: mpsc::Receiver<RemoteHandle<Result<AliceState>>>, alice_swap_handle: mpsc::Receiver<Swap>,
bob_params: BobParams, bob_params: BobParams,
bob_starting_balances: StartingBalances, bob_starting_balances: StartingBalances,
@ -107,7 +106,11 @@ pub struct TestContext {
} }
impl TestContext { impl TestContext {
pub async fn new_swap_as_bob(&mut self) -> (bob::Swap, BobEventLoopJoinHandle) { pub async fn alice_next_swap(&mut self) -> alice::Swap {
self.alice_swap_handle.recv().await.unwrap()
}
pub async fn bob_swap(&mut self) -> (bob::Swap, BobEventLoopJoinHandle) {
let (event_loop, event_loop_handle) = self.bob_params.new_eventloop().unwrap(); let (event_loop, event_loop_handle) = self.bob_params.new_eventloop().unwrap();
let swap = self let swap = self
@ -145,10 +148,7 @@ impl TestContext {
(swap, BobEventLoopJoinHandle(join_handle)) (swap, BobEventLoopJoinHandle(join_handle))
} }
pub async fn assert_alice_redeemed(&mut self) { pub async fn assert_alice_redeemed(&mut self, state: AliceState) {
let swap_handle = self.alice_swap_handle.recv().await.unwrap();
let state = swap_handle.await.unwrap();
assert!(matches!(state, AliceState::BtcRedeemed)); assert!(matches!(state, AliceState::BtcRedeemed));
self.alice_bitcoin_wallet.sync().await.unwrap(); self.alice_bitcoin_wallet.sync().await.unwrap();
@ -175,10 +175,7 @@ impl TestContext {
); );
} }
pub async fn assert_alice_refunded(&mut self) { pub async fn assert_alice_refunded(&mut self, state: AliceState) {
let swap_handle = self.alice_swap_handle.recv().await.unwrap();
let state = swap_handle.await.unwrap();
assert!(matches!(state, AliceState::XmrRefunded)); assert!(matches!(state, AliceState::XmrRefunded));
self.alice_bitcoin_wallet.sync().await.unwrap(); self.alice_bitcoin_wallet.sync().await.unwrap();
@ -313,7 +310,7 @@ impl TestContext {
pub async fn setup_test<T, F, C>(_config: C, testfn: T) pub async fn setup_test<T, F, C>(_config: C, testfn: T)
where where
T: Fn(TestContext) -> F, T: Fn(TestContext) -> F,
F: Future<Output = ()>, F: Future<Output = Result<()>>,
C: GetExecutionParams, C: GetExecutionParams,
{ {
let cli = Cli::default(); let cli = Cli::default();
@ -426,7 +423,7 @@ where
bob_monero_wallet, bob_monero_wallet,
}; };
testfn(test).await; testfn(test).await.unwrap()
} }
fn random_prefix() -> String { fn random_prefix() -> String {