[MM-62178] Fix signaling after WS reconnect in non-RTCD HA

server/cluster_message.go

@@ -13,7 +13,10 @@ import (
 )
 
 type clusterMessage struct {
-	ConnID        string           `json:"conn_id,omitempty"`
+	ConnID string `json:"conn_id,omitempty"`
+	// NewConnID is used by clusterMessageTypeReconnect to inform other nodes
+	// about the new WS connection ID in case it changed.
+	NewConnID     string           `json:"new_conn_id,omitempty"`
 	UserID        string           `json:"user_id,omitempty"`
 	ChannelID     string           `json:"channel_id,omitempty"`
 	CallID        string           `json:"call_id,omitempty"`

server/plugin.go

@@ -178,6 +178,10 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 			close(us.wsReconnectCh)
 			if !us.rtc {
 				delete(p.sessions, us.connID)
+			} else {
+				// If we are the RTC handler for this session, we need to update the connID with the new connection ID in case it changed.
+				// This is needed to correctly send out WS events (e.g. signaling) since we target a specific ConnectionId in the broadcast.
+				us.connID = msg.NewConnID
 			}
 		} else {
 			return fmt.Errorf("session already reconnected, connID=%q", msg.ConnID)
@@ -214,10 +218,8 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 		return nil
 	case clusterMessageTypeSignaling:
 		p.LogDebug("signaling event", "ChannelID", msg.ChannelID, "UserID", msg.UserID, "ConnID", msg.ConnID)
-		p.mut.RLock()
-		us := p.sessions[msg.ConnID]
-		p.mut.RUnlock()
 
+		us := p.getSessionByOriginalID(msg.ConnID)
 		if us == nil {
 			return fmt.Errorf("session doesn't exist, ev=%s, userID=%q, connID=%q, channelID=%q",
 				ev.Id, msg.UserID, msg.ConnID, msg.ChannelID)
@@ -231,7 +233,7 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 			msgType = rtc.ICEMessage
 		}
 		rtcMsg := rtc.Message{
-			SessionID: us.connID,
+			SessionID: us.originalConnID,
 			Type:      msgType,
 			Data:      msg.ClientMessage.Data,
 		}
@@ -241,10 +243,8 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 		}
 	case clusterMessageTypeUserState:
 		p.LogDebug("user state event", "ChannelID", msg.ChannelID, "UserID", msg.UserID, "ConnID", msg.ConnID)
-		p.mut.RLock()
-		us := p.sessions[msg.ConnID]
-		p.mut.RUnlock()
 
+		us := p.getSessionByOriginalID(msg.ConnID)
 		if us == nil {
 			return fmt.Errorf("session doesn't exist, ev=%s, userID=%q, connID=%q, channelID=%q",
 				ev.Id, msg.UserID, msg.ConnID, msg.ChannelID)
@@ -265,7 +265,7 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 		}
 
 		rtcMsg := rtc.Message{
-			SessionID: us.connID,
+			SessionID: us.originalConnID,
 			Type:      msgType,
 			Data:      msg.ClientMessage.Data,
 		}

server/websocket.go

@@ -78,9 +78,7 @@ var (
 	newBatcher = batching.NewBatcher
 )
 
-var (
-	sessionAuthCheckInterval = 10 * time.Second
-)
+var sessionAuthCheckInterval = 10 * time.Second
 
 type CallsClientJoinData struct {
 	ChannelID string
@@ -569,11 +567,10 @@ func (p *Plugin) wsWriter() {
 			if !ok {
 				return
 			}
-			p.mut.RLock()
-			us := p.sessions[msg.SessionID]
-			p.mut.RUnlock()
+
+			us := p.getSessionByOriginalID(msg.SessionID)
 			if us == nil {
-				p.LogError("session should not be nil")
+				p.LogError("failed to get session by originalConnID", "originalConnID", msg.SessionID)
 				continue
 			}
 
@@ -600,7 +597,7 @@ func (p *Plugin) wsWriter() {
 			p.publishWebSocketEvent(wsEventSignal, map[string]interface{}{
 				"data":   string(msg.Data),
 				"connID": msg.SessionID,
-			}, &WebSocketBroadcast{ConnectionID: msg.SessionID, ReliableClusterSend: true})
+			}, &WebSocketBroadcast{ConnectionID: us.connID, ReliableClusterSend: true})
 		case <-p.stopCh:
 			return
 		}
@@ -617,8 +614,8 @@ func (p *Plugin) handleLeave(us *session, userID, connID, channelID, handlerID s
 		// Clearing the previous session since it gets copied over after
 		// successful reconnect.
 		p.mut.Lock()
-		if p.sessions[connID] == us {
-			p.LogDebug("clearing session after reconnect", "userID", userID, "connID", connID, "channelID", channelID)
+		if p.sessions[connID] == us && !us.rtc {
+			p.LogDebug("clearing non-RTC session after reconnect", "userID", userID, "connID", connID, "channelID", channelID)
 			delete(p.sessions, connID)
 		}
 		p.mut.Unlock()
@@ -1062,14 +1059,21 @@ func (p *Plugin) handleReconnect(userID, connID, channelID, originalConnID, prev
 
 	us = newUserSession(userID, channelID, connID, state.Call.ID, rtc)
 	us.originalConnID = originalConnID
+	if p.sessions[originalConnID] != nil {
+		// We need to ensure to clear the original session to avoid potentially tracking it twice in case the ID has changed
+		// and we are the node handling it's RTC counterpart.
+		p.LogDebug("clearing original session after reconnect", "userID", userID, "connID", connID, "originalConnID", originalConnID, "channelID", channelID)
+		delete(p.sessions, originalConnID)
+	}
 	p.sessions[connID] = us
 	p.mut.Unlock()
 
 	if err := p.sendClusterMessage(clusterMessage{
-		ConnID:   prevConnID,
-		UserID:   userID,
-		CallID:   state.Call.ID,
-		SenderID: p.nodeID,
+		ConnID:    prevConnID,
+		NewConnID: connID,
+		UserID:    userID,
+		CallID:    state.Call.ID,
+		SenderID:  p.nodeID,
 	}, clusterMessageTypeReconnect, ""); err != nil {
 		p.LogError(err.Error())
 	}