Restart event loop in tests

Alice was attempting to create a new event loop using the same listen addr as the old one which was still running. This commit aborts the event loop before creating a new one.

swap/src/main.rs

@@ -96,8 +96,7 @@ async fn main() -> Result<()> {
                 Arc::new(monero_wallet),
                 db_path,
                 listen_addr,
-            )
+            );
-            .await;
             let (swap, mut event_loop) =
                 alice_factory.with_init_params(swap_amounts).build().await?;
 
@@ -185,8 +184,7 @@ async fn main() -> Result<()> {
                 Arc::new(monero_wallet),
                 db_path,
                 listen_addr,
-            )
+            );
-            .await;
             let (swap, mut event_loop) = alice_factory.build().await?;
 
             tokio::spawn(async move { event_loop.run().await });

swap/src/protocol/alice.rs

@@ -75,7 +75,7 @@ enum InitParams {
 }
 
 impl Builder {
-    pub async fn new(
+    pub fn new(
         seed: Seed,
         config: Config,
         swap_id: Uuid,
@@ -210,7 +210,7 @@ impl Builder {
         EventLoop::new(
             alice_transport,
             alice_behaviour,
-            self.listen_address.clone(),
+            self.listen_address(),
             self.peer_id,
         )
     }

swap/tests/happy_path.rs

@@ -8,12 +8,12 @@ use tokio::join;
 #[tokio::test]
 async fn happy_path() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, _) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, _) = ctx.new_swap_as_bob().await;
 
         let alice = alice::run(alice_swap);
-
         let bob = bob::run(bob_swap);
+
         let (alice_state, bob_state) = join!(alice, bob);
 
         ctx.assert_alice_redeemed(alice_state.unwrap()).await;

swap/tests/happy_path_restart_alice.rs

@@ -6,8 +6,8 @@ use testutils::alice_run_until::is_encsig_learned;
 #[tokio::test]
 async fn given_alice_restarts_after_encsig_is_learned_resume_swap() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, alice_join_handle) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, _) = ctx.new_swap_as_bob().await;
 
         let bob = bob::run(bob_swap);
         let bob_handle = tokio::spawn(bob);
@@ -17,7 +17,7 @@ async fn given_alice_restarts_after_encsig_is_learned_resume_swap() {
             .unwrap();
         assert!(matches!(alice_state, AliceState::EncSigLearned {..}));
 
-        let alice_swap = ctx.recover_alice_from_db().await;
+        let alice_swap = ctx.stop_and_resume_alice_from_db(alice_join_handle).await;
         assert!(matches!(alice_swap.state, AliceState::EncSigLearned {..}));
 
         let alice_state = alice::run(alice_swap).await.unwrap();

swap/tests/happy_path_restart_bob_after_comm.rs

@@ -6,8 +6,8 @@ use testutils::bob_run_until::is_encsig_sent;
 #[tokio::test]
 async fn given_bob_restarts_after_encsig_is_sent_resume_swap() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, _) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await;
 
         let alice = alice::run(alice_swap);
         let alice_handle = tokio::spawn(alice);
@@ -16,7 +16,7 @@ async fn given_bob_restarts_after_encsig_is_sent_resume_swap() {
 
         assert!(matches!(bob_state, BobState::EncSigSent {..}));
 
-        let bob_swap = ctx.recover_bob_from_db().await;
+        let bob_swap = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
         assert!(matches!(bob_swap.state, BobState::EncSigSent {..}));
 
         let bob_state = bob::run(bob_swap).await.unwrap();

swap/tests/happy_path_restart_bob_after_lock_proof_received.rs

@@ -6,8 +6,8 @@ use testutils::bob_run_until::is_lock_proof_received;
 #[tokio::test]
 async fn given_bob_restarts_after_lock_proof_received_resume_swap() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, _) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await;
 
         let alice_handle = alice::run(alice_swap);
         let alice_swap_handle = tokio::spawn(alice_handle);
@@ -18,8 +18,9 @@ async fn given_bob_restarts_after_lock_proof_received_resume_swap() {
 
         assert!(matches!(bob_state, BobState::XmrLockProofReceived {..}));
 
-        let bob_swap = ctx.recover_bob_from_db().await;
+        let bob_swap = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
-        assert!(matches!(bob_swap.state, BobState::XmrLockProofReceived {..}));
+        assert!(matches!(bob_swap.state, BobState::XmrLockProofReceived
+        {..}));
 
         let bob_state = bob::run(bob_swap).await.unwrap();
 

swap/tests/happy_path_restart_bob_before_comm.rs

@@ -6,8 +6,8 @@ use testutils::bob_run_until::is_xmr_locked;
 #[tokio::test]
 async fn given_bob_restarts_after_xmr_is_locked_resume_swap() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, _) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await;
 
         let alice_handle = alice::run(alice_swap);
         let alice_swap_handle = tokio::spawn(alice_handle);
@@ -16,7 +16,7 @@ async fn given_bob_restarts_after_xmr_is_locked_resume_swap() {
 
         assert!(matches!(bob_state, BobState::XmrLocked {..}));
 
-        let bob_swap = ctx.recover_bob_from_db().await;
+        let bob_swap = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
         assert!(matches!(bob_swap.state, BobState::XmrLocked {..}));
 
         let bob_state = bob::run(bob_swap).await.unwrap();

swap/tests/punish.rs

@@ -8,8 +8,8 @@ use testutils::bob_run_until::is_btc_locked;
 #[tokio::test]
 async fn alice_punishes_if_bob_never_acts_after_fund() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, _) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, bob_join_handle) = ctx.new_swap_as_bob().await;
 
         let alice = alice::run(alice_swap);
         let alice_handle = tokio::spawn(alice);
@@ -23,7 +23,7 @@ async fn alice_punishes_if_bob_never_acts_after_fund() {
 
         // Restart Bob after Alice punished to ensure Bob transitions to
         // punished and does not run indefinitely
-        let bob_swap = ctx.recover_bob_from_db().await;
+        let bob_swap = ctx.stop_and_resume_bob_from_db(bob_join_handle).await;
         assert!(matches!(bob_swap.state, BobState::BtcLocked {..}));
 
         let bob_state = bob::run(bob_swap).await.unwrap();

swap/tests/refund_restart_alice.rs

@@ -8,21 +8,22 @@ use testutils::alice_run_until::is_xmr_locked;
 #[tokio::test]
 async fn given_alice_restarts_after_xmr_is_locked_refund_swap() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, alice_join_handle) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, _) = ctx.new_swap_as_bob().await;
 
         let bob = bob::run(bob_swap);
         let bob_handle = tokio::spawn(bob);
 
         let alice_state = alice::run_until(alice_swap, is_xmr_locked).await.unwrap();
-        assert!(matches!(alice_state, AliceState::XmrLocked {..}));
+        assert!(matches!(alice_state,
+        AliceState::XmrLocked {..}));
 
         // Alice does not act, Bob refunds
         let bob_state = bob_handle.await.unwrap();
         ctx.assert_bob_refunded(bob_state.unwrap()).await;
 
         // Once bob has finished Alice is restarted and refunds as well
-        let alice_swap = ctx.recover_alice_from_db().await;
+        let alice_swap = ctx.stop_and_resume_alice_from_db(alice_join_handle).await;
         assert!(matches!(alice_swap.state, AliceState::XmrLocked {..}));
 
         let alice_state = alice::run(alice_swap).await.unwrap();

swap/tests/refund_restart_alice_cancelled.rs

@@ -9,8 +9,8 @@ use testutils::alice_run_until::is_encsig_learned;
 #[tokio::test]
 async fn given_alice_restarts_after_enc_sig_learned_and_bob_already_cancelled_refund_swap() {
     testutils::setup_test(|mut ctx| async move {
-        let alice_swap = ctx.new_swap_as_alice().await;
+        let (alice_swap, alice_join_handle) = ctx.new_swap_as_alice().await;
-        let bob_swap = ctx.new_swap_as_bob().await;
+        let (bob_swap, _) = ctx.new_swap_as_bob().await;
 
         let bob = bob::run(bob_swap);
         let bob_handle = tokio::spawn(bob);
@@ -25,8 +25,9 @@ async fn given_alice_restarts_after_enc_sig_learned_and_bob_already_cancelled_re
         ctx.assert_bob_refunded(bob_state.unwrap()).await;
 
         // Once bob has finished Alice is restarted and refunds as well
-        let alice_swap = ctx.recover_alice_from_db().await;
+        let alice_swap = ctx.stop_and_resume_alice_from_db(alice_join_handle).await;
-        assert!(matches!(alice_swap.state, AliceState::EncSigLearned {..}));
+        assert!(matches!(alice_swap.state, AliceState::EncSigLearned
+        {..}));
 
         let alice_state = alice::run(alice_swap).await.unwrap();
 

swap/tests/testutils/mod.rs

@@ -14,6 +14,7 @@ use swap::{
 };
 use tempfile::tempdir;
 use testcontainers::{clients::Cli, Container};
+use tokio::task::JoinHandle;
 use tracing_core::dispatcher::DefaultGuard;
 use tracing_log::LogTracer;
 use uuid::Uuid;
@@ -35,7 +36,7 @@ struct AliceParams {
 }
 
 impl AliceParams {
-    pub async fn builder(&self) -> alice::Builder {
+    pub fn builder(&self) -> alice::Builder {
         alice::Builder::new(
             self.seed,
             self.config,
@@ -45,14 +46,14 @@ impl AliceParams {
             self.db_path.clone(),
             self.listen_address.clone(),
         )
-        .await
     }
 
-    async fn peer_id(&self) -> PeerId {
+    fn peer_id(&self) -> PeerId {
-        self.builder().await.peer_id()
+        self.builder().peer_id()
     }
 }
 
+#[derive(Debug, Clone)]
 struct BobParams {
     seed: Seed,
     db_path: PathBuf,
@@ -79,6 +80,10 @@ impl BobParams {
     }
 }
 
+pub struct BobEventLoopJoinHandle(JoinHandle<()>);
+
+pub struct AliceEventLoopJoinHandle(JoinHandle<()>);
+
 pub struct TestContext {
     swap_amounts: SwapAmounts,
 
@@ -94,22 +99,21 @@ pub struct TestContext {
 }
 
 impl TestContext {
-    pub async fn new_swap_as_alice(&mut self) -> alice::Swap {
+    pub async fn new_swap_as_alice(&mut self) -> (alice::Swap, AliceEventLoopJoinHandle) {
         let (swap, mut event_loop) = self
             .alice_params
             .builder()
-            .await
             .with_init_params(self.swap_amounts)
             .build()
             .await
             .unwrap();
 
-        tokio::spawn(async move { event_loop.run().await });
+        let join_handle = tokio::spawn(async move { event_loop.run().await });
 
-        swap
+        (swap, AliceEventLoopJoinHandle(join_handle))
     }
 
-    pub async fn new_swap_as_bob(&mut self) -> bob::Swap {
+    pub async fn new_swap_as_bob(&mut self) -> (bob::Swap, BobEventLoopJoinHandle) {
         let (swap, event_loop) = self
             .bob_params
             .builder()
@@ -118,20 +122,30 @@ impl TestContext {
             .await
             .unwrap();
 
-        tokio::spawn(async move { event_loop.run().await });
+        let join_handle = tokio::spawn(async move { event_loop.run().await });
 
-        swap
+        (swap, BobEventLoopJoinHandle(join_handle))
     }
 
-    pub async fn recover_alice_from_db(&mut self) -> alice::Swap {
+    pub async fn stop_and_resume_alice_from_db(
-        let (swap, mut event_loop) = self.alice_params.builder().await.build().await.unwrap();
+        &mut self,
+        join_handle: AliceEventLoopJoinHandle,
+    ) -> alice::Swap {
+        join_handle.0.abort();
+
+        let (swap, mut event_loop) = self.alice_params.builder().build().await.unwrap();
 
         tokio::spawn(async move { event_loop.run().await });
 
         swap
     }
 
-    pub async fn recover_bob_from_db(&mut self) -> bob::Swap {
+    pub async fn stop_and_resume_bob_from_db(
+        &mut self,
+        join_handle: BobEventLoopJoinHandle,
+    ) -> bob::Swap {
+        join_handle.0.abort();
+
         let (swap, event_loop) = self.bob_params.builder().build().await.unwrap();
 
         tokio::spawn(async move { event_loop.run().await });
@@ -357,7 +371,7 @@ where
         bitcoin_wallet: bob_bitcoin_wallet.clone(),
         monero_wallet: bob_monero_wallet.clone(),
         alice_address: alice_params.listen_address.clone(),
-        alice_peer_id: alice_params.peer_id().await,
+        alice_peer_id: alice_params.peer_id(),
         config,
     };