test: added few taproot integration test case

Author: 0xEgao

src/bin/taker.rs

@@ -200,6 +200,8 @@ fn main() -> Result<(), TakerError> {
                 args.tor_auth,
                 args.zmq,
                 None,
+                #[cfg(feature = "integration-test")]
+                TakerBehavior::Normal,
             )?;
             taproot_taker.recover_from_swap()?;
         }
@@ -214,6 +216,8 @@ fn main() -> Result<(), TakerError> {
                 args.tor_auth,
                 args.zmq,
                 None,
+                #[cfg(feature = "integration-test")]
+                TakerBehavior::Normal,
             )?;
 
             let taproot_swap_params = coinswap::taker::api2::SwapParams {

src/maker/api2.rs

@@ -918,6 +918,20 @@ impl Maker {
                 .contract_tx
                 .compute_txid();
 
+            for (vout, _) in connection_state
+                .incoming_contract
+                .contract_tx
+                .output
+                .iter()
+                .enumerate()
+            {
+                let outpoint = OutPoint {
+                    txid: incoming_txid,
+                    vout: vout as u32,
+                };
+                self.watch_service.unwatch(outpoint);
+            }
+
             // Record the swept coin to track swap balance
             let output_scriptpubkey = spending_tx.output[0].script_pubkey.clone();
             // [TODO] Look into the key value pair later, it shouldn't be both sriptpubkey
@@ -926,6 +940,9 @@ impl Maker {
                 .swept_incoming_swapcoins
                 .insert(output_scriptpubkey.clone(), output_scriptpubkey);
 
+            let mut conn_state = self.ongoing_swap_state.lock()?;
+            conn_state.remove(connection_state.incoming_contract.swap_id.as_ref().unwrap());
+
             wallet.remove_incoming_swapcoin_v2(&incoming_txid);
             log::info!(
                 "[{}] Removed incoming swapcoin {} after successful sweep",

src/maker/server2.rs

@@ -5,7 +5,7 @@
 //! It uses the new message protocol (messages2) and integrates with api2.rs and handlers2.rs.
 //! The server listens at two ports: 6102 for P2P, and 6103 for RPC Client requests.
 
-use bitcoin::Amount;
+use bitcoin::{Amount, OutPoint};
 use bitcoind::bitcoincore_rpc::RpcApi;
 use std::{
     io::ErrorKind,
@@ -474,7 +474,23 @@ fn handle_client_taproot(maker: &Arc<Maker>, stream: &mut TcpStream) -> Result<(
                         .outgoing_contract
                         .contract_tx
                         .compute_txid();
+                    for (vout, _) in connection_state
+                        .outgoing_contract
+                        .contract_tx
+                        .output
+                        .iter()
+                        .enumerate()
+                    {
+                        let outpoint = OutPoint {
+                            txid: outgoing_txid,
+                            vout: vout as u32,
+                        };
+                        maker.watch_service.unwatch(outpoint);
+                    }
                     {
+                        let mut conn_state = maker.ongoing_swap_state.lock()?;
+                        conn_state
+                            .remove(connection_state.outgoing_contract.swap_id.as_ref().unwrap());
                         let mut wallet = maker.wallet.write()?;
                         wallet.remove_outgoing_swapcoin_v2(&outgoing_txid);
                         log::info!(

src/taker/api2.rs

@@ -1,3 +1,6 @@
+use super::error::TakerError;
+#[cfg(feature = "integration-test")]
+use crate::taker::TakerBehavior;
 use crate::{
     protocol::{
         contract2::{
@@ -51,8 +54,6 @@ use std::{
     time::Duration,
 };
 
-use super::error::TakerError;
-
 /// Represents how a taproot contract output was spent
 #[derive(Debug, Clone)]
 pub enum TaprootSpendingPath {
@@ -320,6 +321,8 @@ pub struct Taker {
     ongoing_swap_state: OngoingSwapState,
     data_dir: PathBuf,
     watch_service: WatchService,
+    #[cfg(feature = "integration-test")]
+    behavior: TakerBehavior,
 }
 
 impl Taker {
@@ -332,6 +335,7 @@ impl Taker {
         tor_auth_password: Option<String>,
         zmq_addr: String,
         password: Option<String>,
+        #[cfg(feature = "integration-test")] behavior: TakerBehavior,
     ) -> Result<Taker, TakerError> {
         let data_dir = data_dir.unwrap_or_else(get_taker_dir);
 
@@ -417,6 +421,8 @@ impl Taker {
             ongoing_swap_state: OngoingSwapState::default(),
             data_dir,
             watch_service,
+            #[cfg(feature = "integration-test")]
+            behavior,
         })
     }
 
@@ -487,6 +493,14 @@ impl Taker {
             self.wallet.wait_for_tx_confirmation(tx.compute_txid())?;
         }
 
+        #[cfg(feature = "integration-test")]
+        {
+            if self.behavior == TakerBehavior::DropConnectionAfterFullSetup {
+                log::error!("Dropping Swap Process after full setup");
+                self.recover_from_swap()?;
+                return Ok(None);
+            }
+        }
         match self
             .negotiate_with_makers_and_coordinate_sweep(&outgoing_signed_contract_transactions)
         {
@@ -497,6 +511,7 @@ impl Taker {
 
                 // Store swap state before reset for report generation
                 let prereset_swapstate = self.ongoing_swap_state.clone();
+                self.save_and_reset_swap_round()?;
 
                 // Sync wallet and generate report
                 self.wallet.sync_and_save()?;
@@ -518,6 +533,44 @@ impl Taker {
         }
     }
 
+    /// Save all the finalized swap data and reset the [`OngoingSwapState`].
+    fn save_and_reset_swap_round(&mut self) -> Result<(), TakerError> {
+        // Mark incoiming swapcoins as done
+        let incoming_swapcoin = &self.ongoing_swap_state.incoming_contract;
+        let contract_txid = incoming_swapcoin.contract_tx.compute_txid();
+        for (vout, _) in incoming_swapcoin.contract_tx.output.iter().enumerate() {
+            let outpoint = OutPoint {
+                txid: contract_txid,
+                vout: vout as u32,
+            };
+            self.watch_service.unwatch(outpoint);
+        }
+
+        // Mark outgoing swapcoins as done.
+        let outgoing_swapcoin = &self.ongoing_swap_state.outgoing_contract;
+        let contract_txid = outgoing_swapcoin.contract_tx.compute_txid();
+        for (vout, _) in outgoing_swapcoin.contract_tx.output.iter().enumerate() {
+            let outpoint = OutPoint {
+                txid: contract_txid,
+                vout: vout as u32,
+            };
+            self.watch_service.unwatch(outpoint);
+        }
+
+        self.wallet.sync_no_fail();
+
+        self.wallet.save_to_disk()?;
+
+        self.clear_ongoing_swaps();
+
+        Ok(())
+    }
+
+    /// Clear the [`OngoingSwapState`].
+    fn clear_ongoing_swaps(&mut self) {
+        self.ongoing_swap_state = OngoingSwapState::default();
+    }
+
     /// Generate a swap report after successful completion of a taproot coinswap
     fn generate_swap_report(
         &self,