fix(watchers): align taker fee validation retries with makers

mm2src/mm2_main/src/lp_swap.rs

@@ -151,8 +151,11 @@ pub const TX_HELPER_PREFIX: TopicPrefix = "txhlp";
 pub(crate) const LEGACY_SWAP_TYPE: u8 = 0;
 pub(crate) const MAKER_SWAP_V2_TYPE: u8 = 1;
 pub(crate) const TAKER_SWAP_V2_TYPE: u8 = 2;
-const MAX_STARTED_AT_DIFF: u64 = 60;
 
+pub(crate) const TAKER_FEE_VALIDATION_ATTEMPTS: usize = 6;
+pub(crate) const TAKER_FEE_VALIDATION_RETRY_DELAY: f64 = 10.;
+
+const MAX_STARTED_AT_DIFF: u64 = 60;
 const NEGOTIATE_SEND_INTERVAL: f64 = 30.;
 
 /// If a certain P2P message is not received, swap will be aborted after this time expires.

mm2src/mm2_main/src/lp_swap/maker_swap.rs

@@ -9,7 +9,8 @@ use super::{broadcast_my_swap_status, broadcast_p2p_tx_msg, broadcast_swap_msg_e
             wait_for_maker_payment_conf_until, AtomicSwap, LockedAmount, MySwapInfo, NegotiationDataMsg,
             NegotiationDataV2, NegotiationDataV3, RecoveredSwap, RecoveredSwapAction, SavedSwap, SavedSwapIo,
             SavedTradeFee, SecretHashAlgo, SwapConfirmationsSettings, SwapError, SwapMsg, SwapPubkeys, SwapTxDataMsg,
-            SwapsContext, TransactionIdentifier, INCLUDE_REFUND_FEE, NO_REFUND_FEE, WAIT_CONFIRM_INTERVAL_SEC};
+            SwapsContext, TransactionIdentifier, INCLUDE_REFUND_FEE, NO_REFUND_FEE, TAKER_FEE_VALIDATION_ATTEMPTS,
+            TAKER_FEE_VALIDATION_RETRY_DELAY, WAIT_CONFIRM_INTERVAL_SEC};
 use crate::lp_dispatcher::{DispatcherContext, LpEvents};
 use crate::lp_network::subscribe_to_topic;
 use crate::lp_ordermatch::MakerOrderBuilder;
@@ -771,13 +772,13 @@ impl MakerSwap {
             {
                 Ok(_) => break,
                 Err(err) => {
-                    if attempts >= 6 {
+                    if attempts >= TAKER_FEE_VALIDATION_ATTEMPTS {
                         return Ok((Some(MakerSwapCommand::Finish), vec![
                             MakerSwapEvent::TakerFeeValidateFailed(ERRL!("{}", err).into()),
                         ]));
                     } else {
                         attempts += 1;
-                        Timer::sleep(10.).await;
+                        Timer::sleep(TAKER_FEE_VALIDATION_RETRY_DELAY).await;
                     }
                 },
             };

mm2src/mm2_main/src/lp_swap/swap_watcher.rs

@@ -1,5 +1,6 @@
 use super::{broadcast_p2p_tx_msg, get_payment_locktime, lp_coinfind, taker_payment_spend_deadline, tx_helper_topic,
-            H256Json, SwapsContext, WAIT_CONFIRM_INTERVAL_SEC};
+            H256Json, SwapsContext, TAKER_FEE_VALIDATION_ATTEMPTS, TAKER_FEE_VALIDATION_RETRY_DELAY,
+            WAIT_CONFIRM_INTERVAL_SEC};
 use crate::lp_network::{P2PRequestError, P2PRequestResult};
 
 use crate::MmError;
@@ -181,24 +182,31 @@ impl State for ValidateTakerFee {
 
     async fn on_changed(self: Box<Self>, watcher_ctx: &mut WatcherStateMachine) -> StateResult<WatcherStateMachine> {
         debug!("Watcher validate taker fee");
-        let validated_f = watcher_ctx
-            .taker_coin
-            .watcher_validate_taker_fee(WatcherValidateTakerFeeInput {
-                taker_fee_hash: watcher_ctx.data.taker_fee_hash.clone(),
-                sender_pubkey: watcher_ctx.verified_pub.clone(),
-                min_block_number: watcher_ctx.data.taker_coin_start_block,
-                fee_addr: DEX_FEE_ADDR_RAW_PUBKEY.clone(),
-                lock_duration: watcher_ctx.data.lock_duration,
-            })
-            .compat();
-
-        if let Err(err) = validated_f.await {
-            return Self::change_state(Stopped::from_reason(StopReason::Error(
-                WatcherError::InvalidTakerFee(format!("{:?}", err)).into(),
-            )));
-        };
 
-        Self::change_state(ValidateTakerPayment {})
+        let validation_result = retry_on_err!(async {
+            watcher_ctx
+                .taker_coin
+                .watcher_validate_taker_fee(WatcherValidateTakerFeeInput {
+                    taker_fee_hash: watcher_ctx.data.taker_fee_hash.clone(),
+                    sender_pubkey: watcher_ctx.verified_pub.clone(),
+                    min_block_number: watcher_ctx.data.taker_coin_start_block,
+                    fee_addr: DEX_FEE_ADDR_RAW_PUBKEY.clone(),
+                    lock_duration: watcher_ctx.data.lock_duration,
+                })
+                .compat()
+                .await
+        })
+        .repeat_every_secs(TAKER_FEE_VALIDATION_RETRY_DELAY)
+        .attempts(TAKER_FEE_VALIDATION_ATTEMPTS)
+        .inspect_err(|e| error!("Error validating taker fee: {}", e))
+        .await;
+
+        match validation_result {
+            Ok(_) => Self::change_state(ValidateTakerPayment {}),
+            Err(repeat_err) => Self::change_state(Stopped::from_reason(StopReason::Error(
+                WatcherError::InvalidTakerFee(repeat_err.to_string()).into(),
+            ))),
+        }
     }
 }