wip

gateway/src/node/client_handling/active_clients.rs

@@ -6,8 +6,17 @@ use dashmap::DashMap;
 use nym_sphinx::DestinationAddressBytes;
 use std::sync::Arc;
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum RecentlyActive {
+    Yes,
+    YesButReset,
+    No,
+}
+
 #[derive(Clone)]
-pub(crate) struct ActiveClientsStore(Arc<DashMap<DestinationAddressBytes, MixMessageSender>>);
+pub(crate) struct ActiveClientsStore(
+    Arc<DashMap<DestinationAddressBytes, (MixMessageSender, RecentlyActive)>>,
+);
 
 impl ActiveClientsStore {
     /// Creates new instance of `ActiveClientsStore` to store in-memory handles to all currently connected clients.
@@ -23,7 +32,7 @@ impl ActiveClientsStore {
     /// * `client`: address of the client for which to obtain the handle.
     pub(crate) fn get(&self, client: DestinationAddressBytes) -> Option<MixMessageSender> {
         let entry = self.0.get(&client)?;
-        let handle = entry.value();
+        let handle = &entry.value().0;
 
         // if the entry is stale, remove it from the map
         // if handle.is_valid() {
@@ -37,6 +46,38 @@ impl ActiveClientsStore {
         }
     }
 
+    pub(crate) fn get_is_recently_active(&self, client: DestinationAddressBytes) -> Option<RecentlyActive> {
+        let entry = self.0.get(&client)?;
+        let is_recently_active = &entry.value().1;
+
+        // if the entry is stale, remove it from the map
+        // if handle.is_valid() {
+        if !entry.value().0.is_closed() {
+            Some(is_recently_active.clone())
+        } else {
+            // drop the reference to the map to prevent deadlocks
+            drop(entry);
+            self.0.remove(&client);
+            None
+        }
+    }
+
+    pub(crate) fn register_activity(&self, client: DestinationAddressBytes) {
+        if let Some(mut entry) = self.0.get_mut(&client) {
+            entry.value_mut().1 = RecentlyActive::Yes;
+        }
+    }
+
+    pub(crate) fn reset_activity(&self, client: DestinationAddressBytes) {
+        if let Some(mut entry) = self.0.get_mut(&client) {
+            if entry.value_mut().1 == RecentlyActive::Yes {
+                entry.value_mut().1 = RecentlyActive::YesButReset;
+            } else if entry.value_mut().1 == RecentlyActive::YesButReset {
+                entry.value_mut().1 = RecentlyActive::No;
+            }
+        }
+    }
+
     /// Indicates particular client has disconnected from the gateway and its handle should get removed.
     ///
     /// # Arguments
@@ -53,11 +94,12 @@ impl ActiveClientsStore {
     /// * `client`: address of the client for which to insert the handle.
     /// * `handle`: the sender channel for all mix packets to be pushed back onto the websocket
     pub(crate) fn insert(&self, client: DestinationAddressBytes, handle: MixMessageSender) {
-        self.0.insert(client, handle);
+        self.0.insert(client, (handle, RecentlyActive::Yes));
     }
 
     /// Get number of active clients in store
     pub(crate) fn size(&self) -> usize {
         self.0.len()
     }
+
 }

gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs

@@ -135,6 +135,12 @@ where
         }
     }
 
+    fn register_activity(&self) {
+        self.inner
+            .active_clients_store
+            .register_activity(self.client.address)
+    }
+
     /// Explicitly removes handle from the global store.
     fn disconnect(self) {
         self.inner

gateway/src/node/client_handling/websocket/connection_handler/fresh.rs

@@ -1,7 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::node::client_handling::active_clients::ActiveClientsStore;
+use crate::node::client_handling::active_clients::{ActiveClientsStore, RecentlyActive};
 use crate::node::client_handling::websocket::connection_handler::coconut::CoconutVerifier;
 use crate::node::client_handling::websocket::connection_handler::{
     AuthenticatedHandler, ClientDetails, InitialAuthResult, SocketStream,
@@ -420,8 +420,28 @@ where
 
         if self.active_clients_store.get(address).is_some() {
             println!("duplicate connection when authenticating");
+
+            let msg = Message::Ping("duplicate connection".to_string().into_bytes());
+            if let Err(err) = self.send_websocket_message(msg).await {
+                warn!(
+                    "Failed to send message over websocket: {err}. \
+                    Assuming the connection is dead.",
+                );
+                println!("disconnecting");
+                self.active_clients_store.disconnect(address);
+            }
+
+            println!("waiting..");
+            let msg = self.read_websocket_message().await;
+            dbg!(&msg);
+
+            return Err(InitialAuthenticationError::DuplicateConnection);
+            // if self.active_clients_store.get_is_recently_active(address) == Some(RecentlyActive::No) {
+            // println!("disconnecting");
+            // self.active_clients_store.disconnect(address)
+            // } else {
             // return Err(InitialAuthenticationError::DuplicateConnection);
-            self.active_clients_store.disconnect(address)
+            // }
         }
 
         let shared_keys = self