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

mm2src/coins/eth.rs

@@ -162,7 +162,7 @@ pub mod erc20;
 use erc20::get_token_decimals;
 
 pub(crate) mod eth_swap_v2;
-use eth_swap_v2::{EthPaymentType, PaymentMethod};
+use eth_swap_v2::{extract_id_from_tx_data, EthPaymentType, PaymentMethod, SpendTxSearchParams};
 
 /// https://github.com/artemii235/etomic-swap/blob/master/contracts/EtomicSwap.sol
 /// Dev chain (195.201.137.5:8565) contract address: 0x83965C539899cC0F918552e5A26915de40ee8852
@@ -2587,73 +2587,27 @@ impl MarketCoinOps for EthCoin {
             },
         };
 
-        let payment_func = try_tx_s!(SWAP_CONTRACT.function(&func_name));
-        let decoded = try_tx_s!(decode_contract_call(payment_func, tx.unsigned().data()));
-        let id = match decoded.first() {
-            Some(Token::FixedBytes(bytes)) => bytes.clone(),
-            invalid_token => {
-                return Err(TransactionErr::Plain(ERRL!(
-                    "Expected Token::FixedBytes, got {:?}",
-                    invalid_token
-                )))
-            },
-        };
-
-        loop {
-            if now_sec() > args.wait_until {
-                return TX_PLAIN_ERR!(
-                    "Waited too long until {} for transaction {:?} to be spent ",
-                    args.wait_until,
-                    tx,
-                );
-            }
-
-            let current_block = match self.current_block().compat().await {
-                Ok(b) => b,
-                Err(e) => {
-                    error!("Error getting block number: {}", e);
-                    Timer::sleep(5.).await;
-                    continue;
-                },
-            };
-
-            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;
-                },
-            };
+        let id = try_tx_s!(extract_id_from_tx_data(tx.unsigned().data(), &SWAP_CONTRACT, &func_name).await);
 
-            let found = events.iter().find(|event| &event.data.0[..32] == id.as_slice());
-
-            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;
-                        },
-                        Err(e) => {
-                            error!("Get tx {} error: {}", tx_hash, e);
-                            Timer::sleep(args.check_every).await;
-                            continue;
-                        },
-                    };
-
-                    return Ok(TransactionEnum::from(try_tx_s!(signed_tx_from_web3_tx(transaction))));
-                }
-            }
+        let find_params = SpendTxSearchParams {
+            swap_contract_address,
+            event_name: "ReceiverSpent",
+            abi_contract: &SWAP_CONTRACT,
+            swap_id: &try_tx_s!(id.as_slice().try_into()),
+            from_block: args.from_block,
+            wait_until: args.wait_until,
+            check_every: args.check_every,
+        };
+        let tx_hash = self
+            .find_transaction_hash_by_event(find_params)
+            .await
+            .map_err(|e| TransactionErr::Plain(e.get_inner().to_string()))?;
 
-            Timer::sleep(5.).await;
-        }
+        let spend_tx = self
+            .wait_for_spend_transaction(tx_hash, args.wait_until, args.check_every)
+            .await
+            .map_err(|e| TransactionErr::Plain(e.get_inner().to_string()))?;
+        Ok(TransactionEnum::from(spend_tx))
     }
 
     fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TxMarshalingErr>> {
@@ -4537,7 +4491,9 @@ impl EthCoin {
                     let function = ERC20_CONTRACT.function("balanceOf")?;
                     let data = function.encode_input(&[Token::Address(address)])?;
 
-                    let res = coin.call_request(address, *token_addr, None, Some(data.into())).await?;
+                    let res = coin
+                        .call_request(address, *token_addr, None, Some(data.into()), BlockNumber::Latest)
+                        .await?;
                     let decoded = function.decode_output(&res.0)?;
                     match decoded[0] {
                         Token::Uint(number) => Ok(number),
@@ -4601,7 +4557,7 @@ impl EthCoin {
         let function = ERC20_CONTRACT.function("balanceOf")?;
         let data = function.encode_input(&[Token::Address(address)])?;
         let res = self
-            .call_request(address, token_address, None, Some(data.into()))
+            .call_request(address, token_address, None, Some(data.into()), BlockNumber::Latest)
             .await?;
         let decoded = function.decode_output(&res.0)?;
 
@@ -4628,7 +4584,7 @@ impl EthCoin {
                 let my_address = self.derivation_method.single_addr_or_err().await?;
                 let data = function.encode_input(&[Token::Address(my_address), Token::Uint(token_id_u256)])?;
                 let result = self
-                    .call_request(my_address, token_addr, None, Some(data.into()))
+                    .call_request(my_address, token_addr, None, Some(data.into()), BlockNumber::Latest)
                     .await?;
                 let decoded = function.decode_output(&result.0)?;
                 match decoded[0] {
@@ -4659,7 +4615,7 @@ impl EthCoin {
                 let data = function.encode_input(&[Token::Uint(token_id_u256)])?;
                 let my_address = self.derivation_method.single_addr_or_err().await?;
                 let result = self
-                    .call_request(my_address, token_addr, None, Some(data.into()))
+                    .call_request(my_address, token_addr, None, Some(data.into()), BlockNumber::Latest)
                     .await?;
                 let decoded = function.decode_output(&result.0)?;
                 match decoded[0] {
@@ -4737,6 +4693,7 @@ impl EthCoin {
         to: Address,
         value: Option<U256>,
         data: Option<Bytes>,
+        block_number: BlockNumber,
     ) -> Result<Bytes, web3::Error> {
         let request = CallRequest {
             from: Some(from),
@@ -4748,7 +4705,7 @@ impl EthCoin {
             ..CallRequest::default()
         };
 
-        self.call(request, Some(BlockId::Number(BlockNumber::Latest))).await
+        self.call(request, Some(BlockId::Number(block_number))).await
     }
 
     pub fn allowance(&self, spender: Address) -> Web3RpcFut<U256> {
@@ -4764,7 +4721,7 @@ impl EthCoin {
                     let data = function.encode_input(&[Token::Address(my_address), Token::Address(spender)])?;
 
                     let res = coin
-                        .call_request(my_address, *token_addr, None, Some(data.into()))
+                        .call_request(my_address, *token_addr, None, Some(data.into()), BlockNumber::Latest)
                         .await?;
                     let decoded = function.decode_output(&res.0)?;
 
@@ -4864,25 +4821,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<()> {
@@ -5192,9 +5154,16 @@ impl EthCoin {
                 .single_addr_or_err()
                 .await
                 .map_err(|e| ERRL!("{}", e))?;
-            coin.call_request(my_address, swap_contract_address, None, Some(data.into()))
-                .await
-                .map_err(|e| ERRL!("{}", e))
+            coin.call_request(
+                my_address,
+                swap_contract_address,
+                None,
+                Some(data.into()),
+                // TODO worth reviewing places where we could use BlockNumber::Pending
+                BlockNumber::Latest,
+            )
+            .await
+            .map_err(|e| ERRL!("{}", e))
         };
 
         Box::new(fut.boxed().compat().and_then(move |bytes| {
@@ -5244,10 +5213,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 {
@@ -7005,7 +6980,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()))
     }
 
@@ -7347,14 +7325,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<[u8; 32], String> {
+        self.extract_secret_v2_impl(spend_tx).await
     }
 }