fix(tpu-v2): fix tpu-v2 wait for payment spend and extract secret

mm2src/coins/eth.rs

@@ -2610,6 +2610,7 @@ impl MarketCoinOps for EthCoin {
             },
         };
 
+        let mut tx_hash: Option<H256> = None;
         loop {
             if now_sec() > args.wait_until {
                 return TX_PLAIN_ERR!(
@@ -2628,42 +2629,61 @@ impl MarketCoinOps for EthCoin {
                 },
             };
 
-            let events = match self
-                .spend_events(swap_contract_address, args.from_block, current_block)
-                .compat()
-                .await
-            {
-                Ok(ev) => ev,
-                Err(e) => {
-                    error!("Error getting spend events: {}", e);
-                    Timer::sleep(5.).await;
-                    continue;
-                },
-            };
+            if tx_hash.is_none() {
+                let mut next_from_block = args.from_block;
 
-            let found = events.iter().find(|event| &event.data.0[..32] == id.as_slice());
+                // Split the range into windows of size logs_block_range
+                while next_from_block <= current_block {
+                    let to_block = std::cmp::min(next_from_block + self.logs_block_range - 1, current_block);
 
-            if let Some(event) = found {
-                if let Some(tx_hash) = event.transaction_hash {
-                    let transaction = match self.transaction(TransactionId::Hash(tx_hash)).await {
-                        Ok(Some(t)) => t,
-                        Ok(None) => {
-                            info!("Tx {} not found yet", tx_hash);
-                            Timer::sleep(args.check_every).await;
-                            continue;
-                        },
+                    let events = match self
+                        .events_from_block(
+                            swap_contract_address,
+                            "ReceiverSpent",
+                            next_from_block,
+                            Some(to_block),
+                            &SWAP_CONTRACT,
+                        )
+                        .await
+                    {
+                        Ok(ev) => ev,
                         Err(e) => {
-                            error!("Get tx {} error: {}", tx_hash, e);
-                            Timer::sleep(args.check_every).await;
+                            error!(
+                                "Error getting spend events from {} to {} block: {}",
+                                next_from_block, to_block, e
+                            );
+                            Timer::sleep(5.).await;
+                            next_from_block += self.logs_block_range;
                             continue;
                         },
                     };
 
-                    return Ok(TransactionEnum::from(try_tx_s!(signed_tx_from_web3_tx(transaction))));
+                    // Check if any event matches the SWAP ID
+                    if let Some(found_event) = events.iter().find(|event| &event.data.0[..32] == id.as_slice()) {
+                        if let Some(hash) = found_event.transaction_hash {
+                            // Store tx_hash to skip fetching events in the next iteration if "eth_getTransactionByHash" is unsuccessful
+                            tx_hash = Some(hash);
+                            break;
+                        }
+                    }
+
+                    // Move to the next block range window
+                    next_from_block += self.logs_block_range;
                 }
             }
 
-            Timer::sleep(5.).await;
+            // Proceed getting spend transaction if we have a tx_hash
+            if let Some(tx_hash) = tx_hash {
+                match self.transaction(TransactionId::Hash(tx_hash)).await {
+                    Ok(Some(t)) => {
+                        return Ok(TransactionEnum::from(try_tx_s!(signed_tx_from_web3_tx(t))));
+                    },
+                    Ok(None) => info!("Tx {} not found yet", tx_hash),
+                    Err(e) => error!("Get tx {} error: {}", tx_hash, e),
+                };
+                Timer::sleep(args.check_every).await;
+                continue;
+            }
         }
     }
 
@@ -4875,25 +4895,30 @@ impl EthCoin {
         Box::new(fut.boxed().compat())
     }
 
-    /// Gets `ReceiverSpent` events from etomic swap smart contract since `from_block`
-    fn spend_events(
+    /// Returns events from `from_block` to `to_block` or current `latest` block.
+    /// According to ["eth_getLogs" doc](https://docs.infura.io/api/networks/ethereum/json-rpc-methods/eth_getlogs) `toBlock` is optional, default is "latest".
+    async fn events_from_block(
         &self,
         swap_contract_address: Address,
+        event_name: &str,
         from_block: u64,
-        to_block: u64,
-    ) -> Box<dyn Future<Item = Vec<Log>, Error = String> + Send> {
-        let contract_event = try_fus!(SWAP_CONTRACT.event("ReceiverSpent"));
-        let filter = FilterBuilder::default()
+        to_block: Option<u64>,
+        swap_contract: &Contract,
+    ) -> MmResult<Vec<Log>, FindPaymentSpendError> {
+        let contract_event = swap_contract.event(event_name)?;
+        let mut filter_builder = FilterBuilder::default()
             .topics(Some(vec![contract_event.signature()]), None, None, None)
             .from_block(BlockNumber::Number(from_block.into()))
-            .to_block(BlockNumber::Number(to_block.into()))
-            .address(vec![swap_contract_address])
-            .build();
-
-        let coin = self.clone();
-
-        let fut = async move { coin.logs(filter).await.map_err(|e| ERRL!("{}", e)) };
-        Box::new(fut.boxed().compat())
+            .address(vec![swap_contract_address]);
+        if let Some(block) = to_block {
+            filter_builder = filter_builder.to_block(BlockNumber::Number(block.into()));
+        }
+        let filter = filter_builder.build();
+        let events_logs = self
+            .logs(filter)
+            .await
+            .map_err(|e| FindPaymentSpendError::Transport(e.to_string()))?;
+        Ok(events_logs)
     }
 
     fn validate_payment(&self, input: ValidatePaymentInput) -> ValidatePaymentFut<()> {
@@ -5255,10 +5280,16 @@ impl EthCoin {
             let to_block = current_block.min(from_block + self.logs_block_range);
 
             let spend_events = try_s!(
-                self.spend_events(swap_contract_address, from_block, to_block)
-                    .compat()
-                    .await
+                self.events_from_block(
+                    swap_contract_address,
+                    "ReceiverSpent",
+                    from_block,
+                    Some(to_block),
+                    &SWAP_CONTRACT
+                )
+                .await
             );
+
             let found = spend_events.iter().find(|event| &event.data.0[..32] == id.as_slice());
 
             if let Some(event) = found {
@@ -7010,7 +7041,10 @@ impl ParseCoinAssocTypes for EthCoin {
         }
     }
 
+    fn addr_to_string(&self, address: &Self::Address) -> String { eth_addr_to_hex(address) }
+
     fn parse_address(&self, address: &str) -> Result<Self::Address, Self::AddressParseError> {
+        // crate `Address::from_str` supports both address variants with and without `0x` prefix
         Address::from_str(address).map_to_mm(|e| EthAssocTypesError::InvalidHexString(e.to_string()))
     }
 
@@ -7352,14 +7386,19 @@ impl TakerCoinSwapOpsV2 for EthCoin {
         self.sign_and_broadcast_taker_payment_spend_impl(gen_args, secret).await
     }
 
-    /// Wrapper for [EthCoin::wait_for_taker_payment_spend_impl]
-    async fn wait_for_taker_payment_spend(
+    /// Wrapper for [EthCoin::find_taker_payment_spend_tx_impl]
+    async fn find_taker_payment_spend_tx(
         &self,
         taker_payment: &Self::Tx,
-        _from_block: u64,
+        from_block: u64,
         wait_until: u64,
-    ) -> MmResult<Self::Tx, WaitForPaymentSpendError> {
-        self.wait_for_taker_payment_spend_impl(taker_payment, wait_until).await
+    ) -> MmResult<Self::Tx, FindPaymentSpendError> {
+        self.find_taker_payment_spend_tx_impl(taker_payment, from_block, wait_until, 10.)
+            .await
+    }
+
+    async fn extract_secret_v2(&self, _secret_hash: &[u8], spend_tx: &Self::Tx) -> Result<Vec<u8>, String> {
+        self.extract_secret_v2_impl(spend_tx).await
     }
 }
 

mm2src/coins/eth/eth_swap_v2/eth_taker_swap_v2.rs

@@ -1,12 +1,13 @@
 use super::{check_decoded_length, validate_amount, validate_from_to_and_status, validate_payment_state,
             EthPaymentType, PaymentMethod, PrepareTxDataError, ZERO_VALUE};
-use crate::eth::{decode_contract_call, get_function_input_data, wei_from_big_decimal, EthCoin, EthCoinType,
-                 ParseCoinAssocTypes, RefundFundingSecretArgs, RefundTakerPaymentArgs, SendTakerFundingArgs,
-                 SignedEthTx, SwapTxTypeWithSecretHash, TakerPaymentStateV2, TransactionErr, ValidateSwapV2TxError,
-                 ValidateSwapV2TxResult, ValidateTakerFundingArgs, TAKER_SWAP_V2};
-use crate::{FundingTxSpend, GenTakerFundingSpendArgs, GenTakerPaymentSpendArgs, SearchForFundingSpendErr,
-            WaitForPaymentSpendError};
+use crate::eth::{decode_contract_call, get_function_input_data, signed_tx_from_web3_tx, wei_from_big_decimal, EthCoin,
+                 EthCoinType, ParseCoinAssocTypes, RefundFundingSecretArgs, RefundTakerPaymentArgs,
+                 SendTakerFundingArgs, SignedEthTx, SwapTxTypeWithSecretHash, TakerPaymentStateV2, TransactionErr,
+                 ValidateSwapV2TxError, ValidateSwapV2TxResult, ValidateTakerFundingArgs, TAKER_SWAP_V2};
+use crate::{FindPaymentSpendError, FundingTxSpend, GenTakerFundingSpendArgs, GenTakerPaymentSpendArgs, MarketCoinOps,
+            SearchForFundingSpendErr};
 use common::executor::Timer;
+use common::log::{error, info};
 use common::now_sec;
 use ethabi::{Function, Token};
 use ethcore_transaction::Action;
@@ -15,7 +16,7 @@ use ethkey::public_to_address;
 use futures::compat::Future01CompatExt;
 use mm2_err_handle::prelude::{MapToMmResult, MmError, MmResult};
 use std::convert::TryInto;
-use web3::types::TransactionId;
+use web3::types::{TransactionId, H256};
 
 const ETH_TAKER_PAYMENT: &str = "ethTakerPayment";
 const ERC20_TAKER_PAYMENT: &str = "erc20TakerPayment";
@@ -457,34 +458,105 @@ impl EthCoin {
         Ok(spend_payment_tx)
     }
 
-    /// Checks that taker payment state is `MakerSpent`.
-    /// Accepts maker spent payment transaction and returns it if payment status is correct.
-    pub(crate) async fn wait_for_taker_payment_spend_impl(
+    pub(crate) async fn find_taker_payment_spend_tx_impl(
         &self,
-        taker_payment: &SignedEthTx,
+        taker_payment: &SignedEthTx, // it's approve_tx in Eth case, as in sign_and_send_taker_funding_spend we return approve_tx tx for it
+        from_block: u64,
         wait_until: u64,
-    ) -> MmResult<SignedEthTx, WaitForPaymentSpendError> {
-        let (decoded, taker_swap_v2_contract) = self
-            .get_decoded_and_swap_contract(taker_payment, "spendTakerPayment")
-            .await?;
+        check_every: f64,
+    ) -> MmResult<SignedEthTx, FindPaymentSpendError> {
+        let taker_swap_v2_contract = self
+            .swap_v2_contracts
+            .ok_or_else(|| {
+                FindPaymentSpendError::Internal("Expected swap_v2_contracts to be Some, but found None".to_string())
+            })?
+            .taker_swap_v2_contract;
+        let approve_func = TAKER_SWAP_V2.function(TAKER_PAYMENT_APPROVE)?;
+        let decoded = decode_contract_call(approve_func, taker_payment.unsigned().data())?;
+        let id = match decoded.first() {
+            Some(Token::FixedBytes(bytes)) => bytes,
+            invalid_token => {
+                return MmError::err(FindPaymentSpendError::InvalidData(format!(
+                    "Expected Token::FixedBytes, got {:?}",
+                    invalid_token
+                )))
+            },
+        };
+        let mut tx_hash: Option<H256> = None;
+        // loop to find maker's spendTakerPayment transaction
         loop {
-            let taker_status = self
-                .payment_status_v2(
-                    taker_swap_v2_contract,
-                    decoded[0].clone(), // id from spendTakerPayment
-                    &TAKER_SWAP_V2,
-                    EthPaymentType::TakerPayments,
-                    TAKER_PAYMENT_STATE_INDEX,
-                )
-                .await?;
-            if taker_status == U256::from(TakerPaymentStateV2::MakerSpent as u8) {
-                return Ok(taker_payment.clone());
-            }
             let now = now_sec();
             if now > wait_until {
-                return MmError::err(WaitForPaymentSpendError::Timeout { wait_until, now });
+                return MmError::err(FindPaymentSpendError::Timeout { wait_until, now });
+            }
+
+            let current_block = match self.current_block().compat().await {
+                Ok(b) => b,
+                Err(e) => {
+                    error!("Error getting block number: {}", e);
+                    Timer::sleep(check_every).await;
+                    continue;
+                },
+            };
+
+            // Skip retrieving events if tx_hash is already found
+            if tx_hash.is_none() {
+                let mut next_from_block = from_block;
+
+                while next_from_block <= current_block {
+                    let to_block = std::cmp::min(next_from_block + self.logs_block_range - 1, current_block);
+
+                    // Fetch TakerPaymentSpent events for the current block range
+                    let events = match self
+                        .events_from_block(
+                            taker_swap_v2_contract,
+                            "TakerPaymentSpent",
+                            next_from_block,
+                            Some(to_block),
+                            &TAKER_SWAP_V2,
+                        )
+                        .await
+                    {
+                        Ok(events) => events,
+                        Err(e) => {
+                            error!(
+                                "Error getting TakerPaymentSpent events from {} to {} block: {}",
+                                next_from_block, to_block, e
+                            );
+                            Timer::sleep(check_every).await;
+                            // Move to next window if there was an error
+                            next_from_block += self.logs_block_range;
+                            continue;
+                        },
+                    };
+
+                    // This is how spent event looks like in EtomicSwapTakerV2: event TakerPaymentSpent(bytes32 id, bytes32 secret).
+                    // Check if any event matches the ID.
+                    if let Some(found_event) = events.into_iter().find(|event| &event.data.0[..32] == id.as_slice()) {
+                        if let Some(hash) = found_event.transaction_hash {
+                            // Store tx_hash to skip fetching events in the next iteration if "eth_getTransactionByHash" is unsuccessful
+                            tx_hash = Some(hash);
+                            break;
+                        }
+                    }
+
+                    next_from_block += self.logs_block_range;
+                }
+            }
+
+            // Proceed to check transaction if we have a tx_hash
+            if let Some(tx_hash) = tx_hash {
+                match self.transaction(TransactionId::Hash(tx_hash)).await {
+                    Ok(Some(t)) => {
+                        let transaction = signed_tx_from_web3_tx(t).map_err(FindPaymentSpendError::Internal)?;
+                        return Ok(transaction);
+                    },
+                    Ok(None) => info!("spendTakerPayment transaction {} not found yet", tx_hash),
+                    Err(e) => error!("Get spendTakerPayment transaction {} error: {}", tx_hash, e),
+                };
+                Timer::sleep(check_every).await;
+                continue;
             }
-            Timer::sleep(10.).await;
         }
     }
 
@@ -692,6 +764,42 @@ impl EthCoin {
 
         Ok((decoded, taker_swap_v2_contract))
     }
+
+    /// Extracts the maker's secret from the input of transaction that calls the `spendTakerPayment` smart contract method.
+    ///
+    ///     function spendTakerPayment(
+    ///         bytes32 id,
+    ///         uint256 amount,
+    ///         uint256 dexFee,
+    ///         address taker,
+    ///         bytes32 takerSecretHash,
+    ///         bytes32 makerSecret,
+    ///         address tokenAddress
+    ///     )
+    pub(crate) async fn extract_secret_v2_impl(&self, spend_tx: &SignedEthTx) -> Result<Vec<u8>, String> {
+        let function = try_s!(TAKER_SWAP_V2.function("spendTakerPayment"));
+        // should be 0xcc90c199
+        let expected_signature = function.short_signature();
+        let signature = &spend_tx.unsigned().data()[0..4];
+        if signature != expected_signature {
+            return ERR!(
+                "Expected 'spendTakerPayment' contract call signature: {:?}, found {:?}",
+                expected_signature,
+                signature
+            );
+        };
+        let decoded = try_s!(decode_contract_call(function, spend_tx.unsigned().data()));
+        if decoded.len() < 7 {
+            return ERR!("Invalid arguments in 'spendTakerPayment' call: {:?}", decoded);
+        }
+        match &decoded[5] {
+            Token::FixedBytes(secret) => Ok(secret.to_vec()),
+            _ => ERR!(
+                "Expected secret to be fixed bytes, but decoded function data is {:?}",
+                decoded
+            ),
+        }
+    }
 }
 
 /// Validation function for ETH taker payment data