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

mm2src/coins/eth/web3_transport/websocket_transport.rs

@@ -97,12 +97,7 @@ impl WebsocketTransport {
         }
     }
 
-    async fn handle_keepalive(
-        &self,
-        wsocket: &mut WebSocketStream,
-        response_notifiers: &mut ExpirableMap<usize, oneshot::Sender<Vec<u8>>>,
-        expires_at: Option<Instant>,
-    ) -> OuterAction {
+    async fn handle_keepalive(&self, wsocket: &mut WebSocketStream, expires_at: Option<Instant>) -> OuterAction {
         const SIMPLE_REQUEST: &str = r#"{"jsonrpc":"2.0","method":"net_version","params":[],"id": 0 }"#;
 
         if let Some(expires_at) = expires_at {
@@ -112,10 +107,7 @@ impl WebsocketTransport {
             }
         }
 
-        // Drop expired response notifier channels
-        response_notifiers.clear_expired_entries();
-
-        let mut should_continue = Default::default();
+        let mut should_continue = false;
         for _ in 0..MAX_ATTEMPTS {
             match wsocket
                 .send(tokio_tungstenite_wasm::Message::Text(SIMPLE_REQUEST.to_string()))
@@ -206,9 +198,6 @@ impl WebsocketTransport {
                     }
 
                     if let Some(id) = inc_event.get("id") {
-                        // just to ensure we don't have outdated entries
-                        response_notifiers.clear_expired_entries();
-
                         let request_id = id.as_u64().unwrap_or_default() as usize;
 
                         if let Some(notifier) = response_notifiers.remove(&request_id) {
@@ -279,7 +268,7 @@ impl WebsocketTransport {
         loop {
             futures_util::select! {
                 _ = keepalive_interval.next().fuse() => {
-                    match self.handle_keepalive(&mut wsocket, &mut response_notifiers, expires_at).await {
+                    match self.handle_keepalive(&mut wsocket, expires_at).await {
                         OuterAction::None => {},
                         OuterAction::Continue => continue,
                         OuterAction::Break => break,

mm2src/common/expirable_map.rs

@@ -5,7 +5,7 @@
 
 use instant::{Duration, Instant};
 use rustc_hash::FxHashMap;
-use std::hash::Hash;
+use std::{collections::BTreeMap, hash::Hash};
 
 #[derive(Clone, Debug)]
 pub struct ExpirableEntry<V> {
@@ -19,48 +19,84 @@ impl<V> ExpirableEntry<V> {
     pub fn update_expiration(&mut self, expires_at: Instant) { self.expires_at = expires_at }
 }
 
-impl<K: Eq + Hash, V> Default for ExpirableMap<K, V> {
+impl<K: Eq + Hash + Copy, V> Default for ExpirableMap<K, V> {
     fn default() -> Self { Self::new() }
 }
 
 /// A map that allows associating values with keys and expiring entries.
-/// It is important to note that this implementation does not automatically
-/// remove any entries; it is the caller's responsibility to invoke `clear_expired_entries`
-/// at specified intervals.
+/// It is important to note that this implementation does not have a background worker to
+/// automatically clear expired entries. Outdated entries are only removed when the control flow
+/// is handed back to the map mutably (i.e. some mutable method of the map is invoked).
 ///
 /// WARNING: This is designed for performance-oriented use-cases utilizing `FxHashMap`
 /// under the hood and is not suitable for cryptographic purposes.
 #[derive(Clone, Debug)]
-pub struct ExpirableMap<K: Eq + Hash, V>(FxHashMap<K, ExpirableEntry<V>>);
+pub struct ExpirableMap<K: Eq + Hash + Copy, V> {
+    map: FxHashMap<K, ExpirableEntry<V>>,
+    /// A sorted inverse map from expiration times to keys to speed up expired entries clearing.
+    expiries: BTreeMap<Instant, K>,
+}
 
-impl<K: Eq + Hash, V> ExpirableMap<K, V> {
+impl<K: Eq + Hash + Copy, V> ExpirableMap<K, V> {
     /// Creates a new empty `ExpirableMap`
     #[inline]
-    pub fn new() -> Self { Self(FxHashMap::default()) }
+    pub fn new() -> Self {
+        Self {
+            map: FxHashMap::default(),
+            expiries: BTreeMap::new(),
+        }
+    }
+
+    /// Returns the associated value if present and not expired.
+    #[inline]
+    pub fn get(&self, k: &K) -> Option<&V> {
+        self.map
+            .get(k)
+            .filter(|v| v.expires_at > Instant::now())
+            .map(|v| &v.value)
+    }
 
-    /// Returns the associated value if present.
+    /// Removes a key-value pair from the map and returns the associated value if present and not expired.
     #[inline]
-    pub fn get(&mut self, k: &K) -> Option<&V> { self.0.get(k).map(|v| &v.value) }
+    pub fn remove(&mut self, k: &K) -> Option<V> {
+        self.map.remove(k).filter(|v| v.expires_at > Instant::now()).map(|v| {
+            self.expiries.remove(&v.expires_at);
+            v.value
+        })
+    }
 
     /// Inserts a key-value pair with an expiration duration.
     ///
     /// If a value already exists for the given key, it will be updated and then
     /// the old one will be returned.
     pub fn insert(&mut self, k: K, v: V, exp: Duration) -> Option<V> {
-        let entry = ExpirableEntry {
-            expires_at: Instant::now() + exp,
-            value: v,
-        };
-
-        self.0.insert(k, entry).map(|v| v.value)
+        self.clear_expired_entries();
+        let expires_at = Instant::now() + exp;
+        let entry = ExpirableEntry { expires_at, value: v };
+        self.expiries.insert(expires_at, k);
+        self.map.insert(k, entry).map(|v| v.value)
     }
 
     /// Removes expired entries from the map.
-    pub fn clear_expired_entries(&mut self) { self.0.retain(|_k, v| Instant::now() < v.expires_at); }
-
-    /// Removes a key-value pair from the map and returns the associated value if present.
-    #[inline]
-    pub fn remove(&mut self, k: &K) -> Option<V> { self.0.remove(k).map(|v| v.value) }
+    ///
+    /// Iterates through the `expiries` in order, removing entries that have expired.
+    /// Stops at the first non-expired entry, leveraging the sorted nature of `BTreeMap`.
+    fn clear_expired_entries(&mut self) {
+        let now = Instant::now();
+
+        // `pop_first()` is used here as it efficiently removes expired entries.
+        // `first_key_value()` was considered as it wouldn't need re-insertion for
+        // non-expired entries, but it would require an extra remove operation for
+        // each expired entry. `pop_first()` needs only one re-insertion per call,
+        // which is an acceptable trade-off compared to multiple remove operations.
+        while let Some((exp, key)) = self.expiries.pop_first() {
+            if exp > now {
+                self.expiries.insert(exp, key);
+                break;
+            }
+            self.map.remove(&key);
+        }
+    }
 }
 
 #[cfg(any(test, target_arch = "wasm32"))]
@@ -80,8 +116,8 @@ mod tests {
         let exp = Duration::from_secs(1);
 
         // Insert 2 entries with 1 sec expiration time
-        expirable_map.insert("key1".to_string(), value.to_string(), exp);
-        expirable_map.insert("key2".to_string(), value.to_string(), exp);
+        expirable_map.insert("key1", value, exp);
+        expirable_map.insert("key2", value, exp);
 
         // Wait for entries to expire
         Timer::sleep(2.).await;
@@ -90,14 +126,14 @@ mod tests {
         expirable_map.clear_expired_entries();
 
         // We waited for 2 seconds, so we shouldn't have any entry accessible
-        assert_eq!(expirable_map.0.len(), 0);
+        assert_eq!(expirable_map.map.len(), 0);
 
         // Insert 5 entries
-        expirable_map.insert("key1".to_string(), value.to_string(), Duration::from_secs(5));
-        expirable_map.insert("key2".to_string(), value.to_string(), Duration::from_secs(4));
-        expirable_map.insert("key3".to_string(), value.to_string(), Duration::from_secs(7));
-        expirable_map.insert("key4".to_string(), value.to_string(), Duration::from_secs(2));
-        expirable_map.insert("key5".to_string(), value.to_string(), Duration::from_millis(3750));
+        expirable_map.insert("key1", value, Duration::from_secs(5));
+        expirable_map.insert("key2", value, Duration::from_secs(4));
+        expirable_map.insert("key3", value, Duration::from_secs(7));
+        expirable_map.insert("key4", value, Duration::from_secs(2));
+        expirable_map.insert("key5", value, Duration::from_millis(3750));
 
         // Wait 2 seconds to expire some entries
         Timer::sleep(2.).await;
@@ -106,6 +142,6 @@ mod tests {
         expirable_map.clear_expired_entries();
 
         // We waited for 2 seconds, only one entry should expire
-        assert_eq!(expirable_map.0.len(), 4);
+        assert_eq!(expirable_map.map.len(), 4);
     });
 }

mm2src/mm2_core/src/data_asker.rs

@@ -113,8 +113,6 @@ pub async fn send_asked_data_rpc(
     asked_data: SendAskedDataRequest,
 ) -> Result<bool, MmError<SendAskedDataError>> {
     let mut awaiting_asks = ctx.data_asker.awaiting_asks.lock().await;
-    awaiting_asks.clear_expired_entries();
-
     match awaiting_asks.remove(&asked_data.data_id) {
         Some(sender) => {
             sender.send(asked_data.data).map_to_mm(|_| {