(DAL) Fix ws deadlock (#1994)

* Revert "fix: prevent possible race condition (#1992)"

This reverts commit 303cd85.

* fix: unregister in separate thread

* fix: fix deadlock

* fix: remove unnecssary threading

node/pkg/dal/api/controller.go

@@ -63,13 +63,12 @@ func HandleWebsocket(conn *websocket.Conn) {
 		}
 
 		if msg.Method == "SUBSCRIBE" {
-			h.mu.Lock()
-			defer h.mu.Unlock()
+			h.mu.RLock()
 			subscriptions, ok := h.clients[threadSafeClient]
 			if !ok {
 				subscriptions = map[string]bool{}
 			}
-
+			h.mu.RUnlock()
 			valid := []string{}
 
 			for _, param := range msg.Params {
@@ -80,7 +79,9 @@ func HandleWebsocket(conn *websocket.Conn) {
 				subscriptions[symbol] = true
 				valid = append(valid, param)
 			}
+			h.mu.Lock()
 			h.clients[threadSafeClient] = subscriptions
+			h.mu.Unlock()
 			err = stats.InsertWebsocketSubscriptions(*ctx, id, valid)
 			if err != nil {
 				log.Error().Err(err).Msg("failed to insert websocket subscription log")

node/pkg/dal/api/hub.go

@@ -54,27 +54,21 @@ func (c *Hub) handleClientRegistration() {
 }
 
 func (c *Hub) addClient(client *ThreadSafeClient) {
-	c.mu.RLock()
-	_, ok := c.clients[client]
-	c.mu.RUnlock()
-	if ok {
+	c.mu.Lock() // Use write lock for both checking and insertion
+	defer c.mu.Unlock()
+	if _, ok := c.clients[client]; ok {
 		return
 	}
-
-	c.mu.Lock()
-	defer c.mu.Unlock()
 	c.clients[client] = make(map[string]bool)
 }
 
 func (c *Hub) removeClient(client *ThreadSafeClient) {
-	c.mu.RLock()
+	c.mu.Lock() // Use write lock for both checking and removal
+	defer c.mu.Unlock()
 	subscriptions, ok := c.clients[client]
-	c.mu.RUnlock()
 	if !ok {
 		return
 	}
-
-	c.mu.Lock()
 	delete(c.clients, client)
 	for symbol := range subscriptions {
 		delete(subscriptions, symbol)