fix(orders): fix cancel order race condition using time-based cache

mm2src/common/expirable_map.rs

@@ -88,9 +88,8 @@ impl<K: Eq + Hash + Copy, V> ExpirableMap<K, V> {
     /// the old one will be returned.
     pub fn insert(&mut self, k: K, v: V, exp: Duration) -> Option<V> {
         self.clear_expired_entries();
-        let expires_at = Instant::now() + exp;
-        let entry = ExpirableEntry { expires_at, value: v };
-        self.expiries.insert(expires_at, k);
+        let entry = ExpirableEntry::new(v, exp);
+        self.expiries.insert(entry.expires_at, k);
         self.map.insert(k, entry).map(|v| v.value)
     }
 

mm2src/mm2_core/Cargo.toml

@@ -20,6 +20,7 @@ lazy_static = "1.4"
 mm2_err_handle = { path = "../mm2_err_handle" }
 mm2_event_stream = { path = "../mm2_event_stream" }
 mm2_metrics = { path = "../mm2_metrics" }
+mm2-libp2p = { path = "../mm2_p2p", package = "mm2_p2p" }
 primitives = { path = "../mm2_bitcoin/primitives" }
 rand = { version = "0.7", features = ["std", "small_rng", "wasm-bindgen"] }
 serde = "1"

mm2src/mm2_core/src/mm_ctx.rs

@@ -1,3 +1,4 @@
+use crate::data_asker::DataAsker;
 #[cfg(feature = "track-ctx-pointer")]
 use common::executor::Timer;
 use common::log::{self, LogLevel, LogOnError, LogState};
@@ -10,6 +11,7 @@ use futures::lock::Mutex as AsyncMutex;
 use gstuff::{try_s, Constructible, ERR, ERRL};
 use lazy_static::lazy_static;
 use mm2_event_stream::{controller::Controller, Event, EventStreamConfiguration};
+use mm2_libp2p::PeerAddress;
 use mm2_metrics::{MetricsArc, MetricsOps};
 use primitives::hash::H160;
 use rand::Rng;
@@ -23,8 +25,6 @@ use std::future::Future;
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
 
-use crate::data_asker::DataAsker;
-
 cfg_wasm32! {
     use mm2_rpc::wasm_rpc::WasmRpcSender;
     use crate::DbNamespaceId;
@@ -145,7 +145,7 @@ pub struct MmCtx {
     #[cfg(not(target_arch = "wasm32"))]
     pub async_sqlite_connection: Constructible<Arc<AsyncMutex<AsyncConnection>>>,
     /// Links the RPC context to the P2P context to handle health check responses.
-    pub healthcheck_response_handler: AsyncMutex<ExpirableMap<String, oneshot::Sender<()>>>,
+    pub healthcheck_response_handler: AsyncMutex<ExpirableMap<PeerAddress, oneshot::Sender<()>>>,
 }
 
 impl MmCtx {

mm2src/mm2_main/src/lp_healthcheck.rs

@@ -9,12 +9,11 @@ use instant::{Duration, Instant};
 use lazy_static::lazy_static;
 use mm2_core::mm_ctx::MmArc;
 use mm2_err_handle::prelude::MmError;
-use mm2_libp2p::{decode_message, encode_message, pub_sub_topic, Libp2pPublic, TopicPrefix};
+use mm2_libp2p::{decode_message, encode_message, pub_sub_topic, Libp2pPublic, PeerAddress, TopicPrefix};
 use mm2_net::p2p::P2PContext;
 use ser_error_derive::SerializeErrorType;
 use serde::{Deserialize, Serialize};
 use std::convert::TryFrom;
-use std::str::FromStr;
 use std::sync::Mutex;
 
 use crate::lp_network::broadcast_p2p_msg;
@@ -37,71 +36,6 @@ pub(crate) struct HealthcheckMessage {
     data: HealthcheckData,
 }
 
-/// Wrapper of `libp2p::PeerId` with trait additional implementations.
-///
-/// TODO: This should be used as a replacement of `libp2p::PeerId` in the entire project.
-#[derive(Clone, Copy, Debug, Display, PartialEq)]
-pub struct PeerAddress(mm2_libp2p::PeerId);
-
-impl From<mm2_libp2p::PeerId> for PeerAddress {
-    fn from(value: mm2_libp2p::PeerId) -> Self { Self(value) }
-}
-
-impl From<PeerAddress> for mm2_libp2p::PeerId {
-    fn from(value: PeerAddress) -> Self { value.0 }
-}
-
-impl Serialize for PeerAddress {
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: serde::Serializer,
-    {
-        serializer.serialize_str(&self.0.to_string())
-    }
-}
-
-impl<'de> Deserialize<'de> for PeerAddress {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        struct PeerAddressVisitor;
-
-        impl<'de> serde::de::Visitor<'de> for PeerAddressVisitor {
-            type Value = PeerAddress;
-
-            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
-                formatter.write_str("a string representation of peer id.")
-            }
-
-            fn visit_str<E>(self, value: &str) -> Result<PeerAddress, E>
-            where
-                E: serde::de::Error,
-            {
-                if value.len() > 100 {
-                    return Err(serde::de::Error::invalid_length(
-                        value.len(),
-                        &"peer id cannot exceed 100 characters.",
-                    ));
-                }
-
-                Ok(mm2_libp2p::PeerId::from_str(value)
-                    .map_err(serde::de::Error::custom)?
-                    .into())
-            }
-
-            fn visit_string<E>(self, value: String) -> Result<PeerAddress, E>
-            where
-                E: serde::de::Error,
-            {
-                self.visit_str(&value)
-            }
-        }
-
-        deserializer.deserialize_str(PeerAddressVisitor)
-    }
-}
-
 #[derive(Debug, Display)]
 enum SignValidationError {
     #[display(
@@ -331,8 +265,7 @@ pub async fn peer_connection_healthcheck_rpc(
 
     {
         let mut book = ctx.healthcheck_response_handler.lock().await;
-        book.clear_expired_entries();
-        book.insert(target_peer_address.to_string(), tx, address_record_exp);
+        book.insert(target_peer_address, tx, address_record_exp);
     }
 
     broadcast_p2p_msg(
@@ -395,7 +328,7 @@ pub(crate) async fn process_p2p_healthcheck_message(ctx: &MmArc, message: mm2_li
         } else {
             // The requested peer is healthy; signal the response channel.
             let mut response_handler = ctx.healthcheck_response_handler.lock().await;
-            if let Some(tx) = response_handler.remove(&sender_peer.to_string()) {
+            if let Some(tx) = response_handler.remove(&sender_peer) {
                 if tx.send(()).is_err() {
                     log::error!("Result channel isn't present for peer '{sender_peer}'.");
                 };
@@ -408,13 +341,13 @@ pub(crate) async fn process_p2p_healthcheck_message(ctx: &MmArc, message: mm2_li
 
 #[cfg(any(test, target_arch = "wasm32"))]
 mod tests {
-    use std::mem::discriminant;
-
     use super::*;
     use common::cross_test;
     use crypto::CryptoCtx;
     use mm2_libp2p::behaviours::atomicdex::generate_ed25519_keypair;
     use mm2_test_helpers::for_tests::mm_ctx_with_iguana;
+    use std::mem::discriminant;
+    use std::str::FromStr;
 
     common::cfg_wasm32! {
         use wasm_bindgen_test::*;

mm2src/mm2_p2p/src/lib.rs

@@ -6,12 +6,14 @@ mod network;
 mod relay_address;
 mod swarm_runtime;
 
+use derive_more::Display;
 use lazy_static::lazy_static;
 use secp256k1::{Message as SecpMessage, PublicKey as Secp256k1Pubkey, Secp256k1, SecretKey, SignOnly, Signature,
                 VerifyOnly};
 use serde::{de, Deserialize, Serialize, Serializer};
 use sha2::digest::Update;
 use sha2::{Digest, Sha256};
+use std::str::FromStr;
 
 pub use crate::swarm_runtime::SwarmRuntime;
 
@@ -43,6 +45,69 @@ lazy_static! {
     static ref SECP_SIGN: Secp256k1<SignOnly> = Secp256k1::signing_only();
 }
 
+/// Wrapper of `libp2p::PeerId` with trait additional implementations.
+///
+/// TODO: This should be used as a replacement of `libp2p::PeerId` in the entire project.
+#[derive(Clone, Copy, Debug, Display, Eq, Hash, PartialEq)]
+pub struct PeerAddress(PeerId);
+
+impl From<PeerId> for PeerAddress {
+    fn from(value: PeerId) -> Self { Self(value) }
+}
+
+impl From<PeerAddress> for PeerId {
+    fn from(value: PeerAddress) -> Self { value.0 }
+}
+
+impl Serialize for PeerAddress {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.0.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for PeerAddress {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct PeerAddressVisitor;
+
+        impl<'de> serde::de::Visitor<'de> for PeerAddressVisitor {
+            type Value = PeerAddress;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("a string representation of peer id.")
+            }
+
+            fn visit_str<E>(self, value: &str) -> Result<PeerAddress, E>
+            where
+                E: serde::de::Error,
+            {
+                if value.len() > 100 {
+                    return Err(serde::de::Error::invalid_length(
+                        value.len(),
+                        &"peer id cannot exceed 100 characters.",
+                    ));
+                }
+
+                Ok(PeerId::from_str(value).map_err(de::Error::custom)?.into())
+            }
+
+            fn visit_string<E>(self, value: String) -> Result<PeerAddress, E>
+            where
+                E: de::Error,
+            {
+                self.visit_str(&value)
+            }
+        }
+
+        deserializer.deserialize_str(PeerAddressVisitor)
+    }
+}
+
 #[derive(Clone, Copy, Debug)]
 pub enum NetworkInfo {
     /// The in-memory network.