refactor: replace Arc<RwLock<HashMap>> with Arc<DashMap> for attested contracts

Replace Arc<RwLock<HashMap<_, _>>> with Arc<DashMap<_, _>> for better
concurrency and simpler API. DashMap provides lock-free reads and
fine-grained per-shard locking, eliminating the need for manual lock
management and improving scalability under concurrent access.

Changes:
- Update AttestedContractMap type alias to use DashMap
- Refactor all read/write lock patterns to use DashMap's direct methods
- Simplify token cleanup task by using DashMap's retain() method
- Remove lock acquisition code throughout the codebase

Co-authored-by: nacho.d.g <iduartgomez@users.noreply.github.com>

Author: github-actions[bot]

crates/core/src/client_events/websocket.rs

@@ -1,9 +1,11 @@
 use std::{
     collections::{HashMap, VecDeque},
-    sync::{Arc, OnceLock, RwLock},
+    sync::{Arc, OnceLock},
     time::Duration,
 };
 
+use dashmap::DashMap;
+
 use axum::{
     extract::{
         ws::{Message, WebSocket},
@@ -53,7 +55,7 @@ const PARALLELISM: usize = 10; // TODO: get this from config, or whatever optima
 impl WebSocketProxy {
     pub fn create_router(server_routing: Router) -> (Self, Router) {
         // Create a default empty attested contracts map
-        let attested_contracts = Arc::new(RwLock::new(HashMap::new()));
+        let attested_contracts = Arc::new(DashMap::new());
         Self::create_router_with_attested_contracts(server_routing, attested_contracts)
     }
 
@@ -287,17 +289,16 @@ async fn websocket_commands(
     Extension(attested_contracts): Extension<AttestedContractMap>,
 ) -> Response {
     let on_upgrade = move |ws: WebSocket| async move {
-        // Get the data we need and immediately drop the lock
+        // Get the data we need from the DashMap
         let auth_and_instance = if let Some(token) = auth_token.as_ref() {
-            let attested_contracts_read = attested_contracts.read().unwrap();
-
             // Only collect and log map contents when trace is enabled
             if tracing::enabled!(tracing::Level::TRACE) {
-                let map_contents: Vec<_> = attested_contracts_read.keys().cloned().collect();
+                let map_contents: Vec<_> = attested_contracts.iter().map(|e| e.key().clone()).collect();
                 tracing::trace!(?token, "attested_contracts map keys: {:?}", map_contents);
             }
 
-            if let Some((cid, _, _)) = attested_contracts_read.get(token) {
+            if let Some(entry) = attested_contracts.get(token) {
+                let (cid, _, _) = entry.value();
                 tracing::trace!(?token, ?cid, "Found token in attested_contracts map");
                 Some((token.clone(), *cid))
             } else {
@@ -307,7 +308,7 @@ async fn websocket_commands(
         } else {
             tracing::trace!("No auth token provided in WebSocket request");
             None
-        }; // RwLockReadGuard is dropped here
+        };
 
         // Only evaluate auth_and_instance for trace when trace is enabled
         if tracing::enabled!(tracing::Level::TRACE) {

crates/core/src/node/mod.rs

@@ -1334,9 +1334,8 @@ pub async fn run_local_node(
             ClientRequest::DelegateOp(op) => {
                 let attested_contract = token.and_then(|token| {
                     gw.attested_contracts
-                        .read()
-                        .ok()
-                        .and_then(|guard| guard.get(&token).map(|(t, _, _)| *t))
+                        .get(&token)
+                        .map(|entry| entry.value().0)
                 });
                 let op_name = match op {
                     DelegateRequest::RegisterDelegate { .. } => "RegisterDelegate",

crates/core/src/server/http_gateway.rs

@@ -1,8 +1,10 @@
 use std::collections::HashMap;
 use std::net::{IpAddr, SocketAddr};
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::Instant;
 
+use dashmap::DashMap;
+
 use axum::extract::Path;
 use axum::response::IntoResponse;
 use axum::routing::get;
@@ -34,8 +36,7 @@ impl std::ops::Deref for HttpGatewayRequest {
 
 /// Maps authentication tokens to contract instances, client IDs, and last access time.
 /// The Instant tracks when the token was last used to enable time-based expiration.
-pub type AttestedContractMap =
-    Arc<RwLock<HashMap<AuthToken, (ContractInstanceId, ClientId, Instant)>>>;
+pub type AttestedContractMap = Arc<DashMap<AuthToken, (ContractInstanceId, ClientId, Instant)>>;
 
 /// A gateway to access and interact with contracts through an HTTP interface.
 pub(crate) struct HttpGateway {
@@ -47,7 +48,7 @@ pub(crate) struct HttpGateway {
 impl HttpGateway {
     /// Returns the uninitialized axum router to compose with other routing handling or websockets.
     pub fn as_router(socket: &SocketAddr) -> (Self, Router) {
-        let attested_contracts = Arc::new(RwLock::new(HashMap::new()));
+        let attested_contracts = Arc::new(DashMap::new());
         Self::as_router_with_attested_contracts(socket, attested_contracts)
     }
 
@@ -87,8 +88,6 @@ impl ClientEventsProxy for HttpGateway {
                         if let Some((assigned_token, contract)) = assigned_token {
                             let now = Instant::now();
                             self.attested_contracts
-                                .write()
-                                .map_err(|_| ErrorKind::FailedOperation)?
                                 .insert(assigned_token.clone(), (contract, cli_id, now));
                             tracing::debug!(
                                 ?assigned_token,

crates/core/src/server/mod.rs

@@ -13,9 +13,11 @@ pub(crate) mod path_handlers;
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use dashmap::DashMap;
+
 use freenet_stdlib::{
     client_api::{ClientError, ClientRequest, HostResponse},
     prelude::*,
@@ -149,13 +151,15 @@ pub mod local_node {
                         tracing::info!("disconnecting cause: {cause}");
                     }
                     // fixme: token must live for a bit to allow reconnections
-                    if let Ok(mut guard) = gw.attested_contracts.write() {
-                        if let Some(rm_token) = guard
-                            .iter()
-                            .find_map(|(k, (_, eid, _))| (eid == &id).then(|| k.clone()))
-                        {
-                            guard.remove(&rm_token);
-                        }
+                    if let Some(rm_token) = gw
+                        .attested_contracts
+                        .iter()
+                        .find_map(|entry| {
+                            let (k, (_, eid, _)) = entry.pair();
+                            (eid == &id).then(|| k.clone())
+                        })
+                    {
+                        gw.attested_contracts.remove(&rm_token);
                     }
                     continue;
                 }
@@ -209,7 +213,7 @@ pub(crate) async fn serve_gateway_in(config: WebsocketApiConfig) -> (HttpGateway
     let ws_socket = (config.address, config.port).into();
 
     // Create a shared attested_contracts map with token expiration support
-    let attested_contracts: AttestedContractMap = Arc::new(RwLock::new(HashMap::new()));
+    let attested_contracts: AttestedContractMap = Arc::new(DashMap::new());
 
     // Spawn background task to clean up expired tokens
     spawn_token_cleanup_task(attested_contracts.clone());
@@ -242,38 +246,34 @@ fn spawn_token_cleanup_task(attested_contracts: AttestedContractMap) {
             interval.tick().await;
 
             // Clean up expired tokens
-            if let Ok(mut guard) = attested_contracts.write() {
-                let now = Instant::now();
-                let initial_count = guard.len();
-
-                // Remove tokens that haven't been accessed in TOKEN_TTL
-                guard.retain(|token, (contract_id, client_id, last_used)| {
-                    let elapsed = now.duration_since(*last_used);
-                    let should_keep = elapsed < TOKEN_TTL;
-
-                    if !should_keep {
-                        tracing::info!(
-                            ?token,
-                            ?contract_id,
-                            ?client_id,
-                            elapsed_hours = elapsed.as_secs() / 3600,
-                            "Removing expired authentication token"
-                        );
-                    }
+            let now = Instant::now();
+            let initial_count = attested_contracts.len();
 
-                    should_keep
-                });
+            // Remove tokens that haven't been accessed in TOKEN_TTL
+            attested_contracts.retain(|token, (contract_id, client_id, last_used)| {
+                let elapsed = now.duration_since(*last_used);
+                let should_keep = elapsed < TOKEN_TTL;
 
-                let removed_count = initial_count - guard.len();
-                if removed_count > 0 {
-                    tracing::debug!(
-                        removed_count,
-                        remaining_count = guard.len(),
-                        "Token cleanup completed"
+                if !should_keep {
+                    tracing::info!(
+                        ?token,
+                        ?contract_id,
+                        ?client_id,
+                        elapsed_hours = elapsed.as_secs() / 3600,
+                        "Removing expired authentication token"
                     );
                 }
-            } else {
-                tracing::warn!("Failed to acquire write lock for token cleanup");
+
+                should_keep
+            });
+
+            let removed_count = initial_count - attested_contracts.len();
+            if removed_count > 0 {
+                tracing::debug!(
+                    removed_count,
+                    remaining_count = attested_contracts.len(),
+                    "Token cleanup completed"
+                );
             }
         }
     });