add complete coverage on order recovery

Signed-off-by: Onur Özkan <[email protected]>

Author: onur-ozkan

mm2src/mm2_main/src/lp_ordermatch.rs

@@ -384,9 +384,11 @@ async fn process_maker_order_cancelled(
     // is received within the `RECENTLY_CANCELLED_TIMEOUT` timeframe.
     // We do this even if the order is in the order_set, because it could have been added through
     // means other than the order creation message.
-    orderbook
-        .recently_cancelled
-        .insert_expirable(uuid, from_pubkey.clone(), RECENTLY_CANCELLED_TIMEOUT);
+    orderbook.recently_cancelled.insert_expirable(
+        uuid,
+        (from_pubkey.clone(), cancelled_msg.timestamp),
+        RECENTLY_CANCELLED_TIMEOUT,
+    );
     if let Some(order) = orderbook.order_set.get(&uuid) {
         if order.pubkey == from_pubkey {
             orderbook.remove_order_trie_update(uuid);
@@ -1108,7 +1110,7 @@ async fn maker_order_updated_p2p_notify(
 pub(crate) async fn maker_order_cancelled_p2p_notify(ctx: &MmArc, order: &MakerOrder) {
     let message = new_protocol::OrdermatchMessage::MakerOrderCancelled(new_protocol::MakerOrderCancelled {
         uuid: order.uuid.into(),
-        timestamp: now_sec(),
+        timestamp: order.created_at,
         pair_trie_root: H64::default(),
     });
     delete_my_order(ctx, order.uuid, order.p2p_privkey).await;
@@ -2576,7 +2578,7 @@ pub(crate) struct Orderbook {
     /// used to avoid order recreation in case of out-of-order p2p messages,
     /// e.g., when receiving the order cancellation message before the order is created.
     /// Entries are kept for `RECENTLY_CANCELLED_TIMEOUT` seconds.
-    recently_cancelled: TimedMap<Uuid, String>,
+    recently_cancelled: TimedMap<Uuid, (String, u64)>,
     topics_subscribed_to: HashMap<String, OrderbookRequestingState>,
     /// MemoryDB instance to store Patricia Tries data
     memory_db: MemoryDB<Blake2Hasher64>,
@@ -2627,7 +2629,7 @@ impl Orderbook {
 
     fn insert_or_update_order_update_trie(&mut self, order: OrderbookItem) {
         // Ignore the order if it was recently cancelled
-        if self.recently_cancelled.get(&order.uuid) == Some(&order.pubkey) {
+        if self.recently_cancelled.get(&order.uuid) == Some(&(order.pubkey.clone(), order.created_at)) {
             warn!("Maker order {} was recently cancelled, ignoring", order.uuid);
             return;
         }
@@ -3049,11 +3051,13 @@ impl MakerOrdersContext {
             .remove(&uuid)
             .ok_or_else(|| ERRL!("Cannot find {} order to recover.", uuid))?;
 
-        self.orders.insert(uuid, order.clone());
-
         {
             let mut order = order.lock().await;
-            order.updated_at = Some(now_ms());
+
+            order.matches.clear();
+            order.started_swaps.clear();
+            order.created_at = now_ms();
+            order.updated_at = None;
 
             let ctx_arc = MmArc::from_weak(&ctx).ok_or_else(|| ERRL!("This is very unusual."))?;
 
@@ -3075,6 +3079,8 @@ impl MakerOrdersContext {
             self.spawn_balance_loop_if_not_spawned(ctx, order.base.clone(), None);
         }
 
+        self.orders.insert(uuid, order.clone());
+
         Ok(order)
     }
 

mm2src/mm2_main/src/lp_swap.rs

@@ -432,6 +432,8 @@ async fn recv_swap_msg<T>(
 
 /// Includes the grace time we add to the "normal" timeouts
 /// in order to give different and/or heavy communication channels a chance.
+///
+/// TODO: It doesn't seem right to need such thing.
 const BASIC_COMM_TIMEOUT: u64 = 90;
 
 #[cfg(not(any(feature = "custom-swap-locktime", test, feature = "run-docker-tests")))]

mm2src/mm2_main/src/lp_swap/maker_swap.rs

@@ -15,7 +15,8 @@ use crate::lp_network::subscribe_to_topic;
 use crate::lp_ordermatch::MakerOrderBuilder;
 use crate::lp_swap::swap_events::{SwapStatusEvent, SwapStatusStreamer};
 use crate::lp_swap::swap_v2_common::mark_swap_as_finished;
-use crate::lp_swap::{broadcast_swap_message, taker_payment_spend_duration, MAX_STARTED_AT_DIFF};
+use crate::lp_swap::{broadcast_swap_message, taker_payment_spend_duration, MAX_STARTED_AT_DIFF,
+                     NEGOTIATION_TIMEOUT_SEC};
 use coins::lp_price::fetch_swap_coins_price;
 use coins::{CanRefundHtlc, CheckIfMyPaymentSentArgs, ConfirmPaymentInput, DexFee, FeeApproxStage, FoundSwapTxSpend,
             MmCoin, MmCoinEnum, PaymentInstructionArgs, PaymentInstructions, PaymentInstructionsErr,
@@ -594,8 +595,6 @@ impl MakerSwap {
 
         let maker_negotiation_msg = SwapMsg::Negotiation(negotiation_data);
 
-        const NEGOTIATION_TIMEOUT_SEC: u64 = 90;
-
         debug!("Sending maker negotiation data: {:?}", maker_negotiation_msg);
         let send_abort_handle = broadcast_swap_msg_every(
             self.ctx.clone(),

mm2src/mm2_main/src/lp_swap/maker_swap_v2.rs

@@ -1160,6 +1160,13 @@ impl<MakerCoin: MmCoin + MakerCoinSwapOpsV2, TakerCoin: MmCoin + TakerCoinSwapOp
     type StateMachine = MakerSwapStateMachine<MakerCoin, TakerCoin>;
 
     async fn on_changed(self: Box<Self>, state_machine: &mut Self::StateMachine) -> StateResult<Self::StateMachine> {
+        #[cfg(feature = "run-docker-tests")]
+        if std::env::var("ABORT_SWAP_FOR_TEST").is_ok() {
+            debug!("Aborting it intentionally.");
+            let reason = AbortReason::FailedToParseTakerFunding("Intentional swap abort.".to_string());
+            return Self::change_state(Aborted::new(reason), state_machine).await;
+        }
+
         let maker_negotiated_msg = MakerNegotiated {
             negotiated: true,
             reason: None,
@@ -2095,6 +2102,8 @@ impl<MakerCoin: MmCoin + MakerCoinSwapOpsV2, TakerCoin: MmCoin + TakerCoinSwapOp
             covered_error!("{error}");
         }
 
+        info!("{order_uuid} order is recovered");
+
         warn!("Swap {} was aborted with reason {}", state_machine.uuid, self.reason);
     }
 }

mm2src/mm2_main/tests/docker_tests/eth_docker_tests.rs

@@ -2872,7 +2872,6 @@ fn test_v2_eth_eth_kickstart() {
 }
 
 #[test]
-#[ignore]
 fn test_maker_order_recovery_on_tpu() {
     // Initialize swap addresses and configurations
     let swap_addresses = SwapAddresses::init();
@@ -2906,52 +2905,105 @@ fn test_maker_order_recovery_on_tpu() {
         }
     };
 
-    // start Bob and Alice
+    // start Bob, Alice and Onur
     let (_, bob_priv_key) =
         eth_coin_v2_activation_with_random_privkey(&MM_CTX, ETH, &eth_dev_conf(), swap_addresses, false);
     let (_, alice_priv_key) =
         eth_coin_v2_activation_with_random_privkey(&MM_CTX1, ETH1, &eth1_dev_conf(), swap_addresses, false);
+    let (_, onur_priv_key) =
+        eth_coin_v2_activation_with_random_privkey(&MM_CTX1, ETH1, &eth1_dev_conf(), swap_addresses, false);
+
     let coins = json!([eth_dev_conf(), eth1_dev_conf()]);
 
     let bob_conf = Mm2TestConf::seednode_trade_v2(&format!("0x{}", hex::encode(bob_priv_key)), &coins);
-    let mut mm_bob = MarketMakerIt::start(bob_conf.conf.clone(), bob_conf.rpc_password, None).unwrap();
-    let (_bob_dump_log, _bob_dump_dashboard) = mm_dump(&mm_bob.log_path);
-    log!("Bob log path: {}", mm_bob.log_path.display());
+    let mut mm_bob = block_on(MarketMakerIt::start_with_envs(
+        bob_conf.conf.clone(),
+        bob_conf.rpc_password,
+        None,
+        &[("ABORT_SWAP_FOR_TEST", "1")],
+    ))
+    .unwrap();
 
     let alice_conf =
         Mm2TestConf::light_node_trade_v2(&format!("0x{}", hex::encode(alice_priv_key)), &coins, &[&mm_bob
             .ip
             .to_string()]);
     let mut mm_alice = MarketMakerIt::start(alice_conf.conf.clone(), alice_conf.rpc_password, None).unwrap();
-    let (_alice_dump_log, _alice_dump_dashboard) = mm_dump(&mm_alice.log_path);
-    log!("Alice log path: {}", mm_alice.log_path.display());
+
+    let onur_conf = Mm2TestConf::light_node_trade_v2(&format!("0x{}", hex::encode(onur_priv_key)), &coins, &[&mm_bob
+        .ip
+        .to_string()]);
+    let mm_onur = MarketMakerIt::start(onur_conf.conf.clone(), onur_conf.rpc_password, None).unwrap();
 
     // Enable ETH and ETH1 for both Bob and Alice
     enable_coins(&mm_bob, &[ETH, ETH1]);
     enable_coins(&mm_alice, &[ETH, ETH1]);
+    enable_coins(&mm_onur, &[ETH, ETH1]);
 
     let uuids = block_on(start_swaps(&mut mm_bob, &mut mm_alice, &[(ETH, ETH1)], 1.0, 1.0, 77.));
     block_on(mm_bob.wait_for_log(30., |log| {
         log.contains(&format!("Maker swap {} has successfully started", uuids[0]))
     }))
     .unwrap();
 
-    // Stop alice right after swap start
-    block_on(mm_alice.stop()).unwrap();
+    let send_my_orders_rpc = |mm: &MarketMakerIt| -> MyOrdersRpcResult {
+        let rc = block_on(mm.rpc(&json!({
+            "userpass": mm.userpass,
+            "method": "my_orders",
+        })))
+        .unwrap();
+        assert!(rc.0.is_success(), "!my_orders: {}", rc.1);
 
-    // TODO:
-    // Wait T amount of seconds to timeout the swap.
-    // Verify bob order recovery.
+        serde_json::from_str(&rc.1).unwrap()
+    };
 
-    let rc = block_on(mm_bob.rpc(&json!({
-        "userpass": mm_bob.userpass,
-        "method": "my_orders",
-    })))
-    .unwrap();
-    assert!(rc.0.is_success(), "!my_orders: {}", rc.1);
+    let send_orderbook_rpc = |mm: &MarketMakerIt| -> Json {
+        let rc = block_on(mm.rpc(&json!({
+            "userpass": mm.userpass,
+            "method": "orderbook",
+            "mmrpc": "2.0",
+            "params": {
+                "base": "ETH",
+                "rel": "ETH1",
+            },
+        })))
+        .unwrap();
+        assert!(rc.0.is_success(), "!orderbook: {}", rc.1);
+
+        serde_json::from_str(&rc.1).unwrap()
+    };
+
+    let my_orders_rpc = send_my_orders_rpc(&mm_bob);
+    assert!(
+        my_orders_rpc.result.maker_orders.is_empty(),
+        "Maker order must be invisible (locked)."
+    );
+
+    let orderbook_result = send_orderbook_rpc(&mm_onur);
+    let asks = orderbook_result["result"]["asks"].as_array().unwrap();
+    assert!(
+        asks.is_empty(),
+        "Onur ETH/ETH1 orderbook shouldn't see Bob's locked order."
+    );
 
-    let res: MyOrdersRpcResult = serde_json::from_str(&rc.1).unwrap();
-    assert!(!res.result.maker_orders.is_empty(), "Maker order must be recovered.");
+    block_on(mm_bob.wait_for_log(30., |log| log.contains("Aborting it intentionally."))).unwrap();
+
+    // Swap will be aborted. Stop Alice to prevent her from taking the order again.
+    block_on(mm_alice.stop()).unwrap();
+
+    let my_orders_rpc = send_my_orders_rpc(&mm_bob);
+    assert!(
+        !my_orders_rpc.result.maker_orders.is_empty(),
+        "Maker order must be visible again (recovered)."
+    );
+
+    let orderbook_result = send_orderbook_rpc(&mm_onur);
+    let asks = orderbook_result["result"]["asks"].as_array().unwrap();
+    assert_eq!(
+        asks.len(),
+        1,
+        "Onur ETH/ETH1 orderbook must see Bob's recovered order again."
+    );
 }
 
 fn log_swap_status_before_stop(mm: &MarketMakerIt, uuid: &str, role: &str) {