Calls v1: Fix silent handleReconnect failures [MM-70184]

server/websocket.go

@@ -1052,7 +1052,7 @@ func (p *Plugin) handleReconnect(userID, connID, channelID, originalConnID, prev
 		return fmt.Errorf("forbidden")
 	}
 
-	state, err := p.getCallState(channelID, false)
+	state, err := p.getCallState(channelID, true)
 	if err != nil {
 		return err
 	} else if state == nil {
@@ -1293,6 +1293,10 @@ func (p *Plugin) WebSocketMessageHasBeenPosted(connID, userID string, req *model
 			if err := p.handleReconnect(userID, connID, channelID, originalConnID, prevConnID, req.Session.Id); err != nil {
 				p.LogWarn(err.Error(), "userID", userID, "connID", connID,
 					"originalConnID", originalConnID, "prevConnID", prevConnID, "channelID", channelID)
+				p.publishWebSocketEvent(wsEventError, map[string]interface{}{
+					"data":   err.Error(),
+					"connID": connID,
+				}, &WebSocketBroadcast{ConnectionID: connID, ReliableClusterSend: true})
 			}
 		}()
 		return

server/websocket_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -1570,3 +1571,203 @@ func TestHandleClientMsgVideoStats(t *testing.T) {
 		require.False(t, state.sessions[connID].Video, "session video flag must be cleared")
 	})
 }
+
+func TestHandleReconnect(t *testing.T) {
+	mockAPI := &pluginMocks.MockAPI{}
+	mockMetrics := &serverMocks.MockMetrics{}
+
+	p := Plugin{
+		MattermostPlugin: plugin.MattermostPlugin{
+			API: mockAPI,
+		},
+		callsClusterLocks: map[string]*cluster.Mutex{},
+		metrics:           mockMetrics,
+		sessions:          map[string]*session{},
+	}
+
+	store, tearDown := NewTestStore(t)
+	t.Cleanup(tearDown)
+	p.store = store
+
+	// log.go prepends "origin"+value, so handleReconnect's opening LogDebug has 13 args.
+	mockAPI.On("LogDebug", mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything)
+	mockAPI.On("KVSetWithOptions", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
+	mockMetrics.On("ObserveClusterMutexGrabTime", "mutex_call", mock.AnythingOfType("float64"))
+	mockMetrics.On("ObserveClusterMutexLockedTime", "mutex_call", mock.AnythingOfType("float64"))
+	mockMetrics.On("ObserveAppHandlersTime", mock.AnythingOfType("string"), mock.AnythingOfType("float64"))
+
+	channelID := model.NewId()
+	userID := model.NewId()
+	connID := model.NewId()
+	originalConnID := model.NewId()
+	prevConnID := model.NewId()
+
+	t.Run("forbidden", func(t *testing.T) {
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionCreatePost).Return(false).Once()
+		err := p.handleReconnect(userID, connID, channelID, originalConnID, prevConnID, "")
+		require.EqualError(t, err, "forbidden")
+	})
+
+	t.Run("no call ongoing", func(t *testing.T) {
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionCreatePost).Return(true).Once()
+		err := p.handleReconnect(userID, connID, channelID, originalConnID, prevConnID, "")
+		require.EqualError(t, err, "no call ongoing")
+	})
+
+	t.Run("session not found in call state", func(t *testing.T) {
+		defer ResetTestStore(t, p.store)
+
+		callID := model.NewId()
+		require.NoError(t, p.store.CreateCall(&public.Call{
+			ID:        callID,
+			CreateAt:  time.Now().UnixMilli(),
+			ChannelID: channelID,
+			StartAt:   time.Now().UnixMilli(),
+			PostID:    model.NewId(),
+			ThreadID:  model.NewId(),
+			OwnerID:   userID,
+		}))
+
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionCreatePost).Return(true).Once()
+		err := p.handleReconnect(userID, connID, channelID, originalConnID, prevConnID, "")
+		require.EqualError(t, err, "session not found in call state")
+	})
+
+	t.Run("session belongs to different user", func(t *testing.T) {
+		defer ResetTestStore(t, p.store)
+
+		callID := model.NewId()
+		require.NoError(t, p.store.CreateCall(&public.Call{
+			ID:        callID,
+			CreateAt:  time.Now().UnixMilli(),
+			ChannelID: channelID,
+			StartAt:   time.Now().UnixMilli(),
+			PostID:    model.NewId(),
+			ThreadID:  model.NewId(),
+			OwnerID:   userID,
+		}))
+		require.NoError(t, p.store.CreateCallSession(&public.CallSession{
+			ID:     originalConnID,
+			CallID: callID,
+			UserID: model.NewId(), // different user owns this session
+			JoinAt: time.Now().UnixMilli(),
+		}))
+
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionCreatePost).Return(true).Once()
+		err := p.handleReconnect(userID, connID, channelID, originalConnID, prevConnID, "")
+		require.EqualError(t, err, "session not found in call state")
+	})
+
+	t.Run("already reconnected", func(t *testing.T) {
+		defer ResetTestStore(t, p.store)
+
+		callID := model.NewId()
+		require.NoError(t, p.store.CreateCall(&public.Call{
+			ID:        callID,
+			CreateAt:  time.Now().UnixMilli(),
+			ChannelID: channelID,
+			StartAt:   time.Now().UnixMilli(),
+			PostID:    model.NewId(),
+			ThreadID:  model.NewId(),
+			OwnerID:   userID,
+		}))
+		require.NoError(t, p.store.CreateCallSession(&public.CallSession{
+			ID:     originalConnID,
+			CallID: callID,
+			UserID: userID,
+			JoinAt: time.Now().UnixMilli(),
+		}))
+
+		// Session in memory with wsReconnected already set to 1.
+		p.mut.Lock()
+		p.sessions[connID] = &session{
+			userID:    userID,
+			channelID: channelID,
+			connID:    connID,
+			callID:    callID,
+		}
+		atomic.StoreInt32(&p.sessions[connID].wsReconnected, 1)
+		p.mut.Unlock()
+		defer func() {
+			p.mut.Lock()
+			delete(p.sessions, connID)
+			p.mut.Unlock()
+		}()
+
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionCreatePost).Return(true).Once()
+		err := p.handleReconnect(userID, connID, channelID, originalConnID, prevConnID, "")
+		require.EqualError(t, err, "session already reconnected")
+	})
+}
+
+func TestWebSocketMessageHasBeenPostedReconnectError(t *testing.T) {
+	mockAPI := &pluginMocks.MockAPI{}
+	mockMetrics := &serverMocks.MockMetrics{}
+
+	p := Plugin{
+		MattermostPlugin: plugin.MattermostPlugin{
+			API: mockAPI,
+		},
+		callsClusterLocks: map[string]*cluster.Mutex{},
+		metrics:           mockMetrics,
+		sessions:          map[string]*session{},
+	}
+
+	store, tearDown := NewTestStore(t)
+	t.Cleanup(tearDown)
+	p.store = store
+
+	// log.go prepends "origin"+value, so handleReconnect's opening LogDebug has 13 args.
+	mockAPI.On("LogDebug", mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything).Maybe()
+	// LogWarn also prepends "origin"+value (2 extra args).
+	mockAPI.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything).Maybe()
+	// Mutex-related mocks are not exercised in the reconnect error path.
+	mockAPI.On("KVSetWithOptions", mock.Anything, mock.Anything, mock.Anything).Return(true, nil).Maybe()
+	mockMetrics.On("ObserveClusterMutexGrabTime", "mutex_call", mock.AnythingOfType("float64")).Maybe()
+	mockMetrics.On("ObserveClusterMutexLockedTime", "mutex_call", mock.AnythingOfType("float64")).Maybe()
+	mockMetrics.On("ObserveAppHandlersTime", mock.AnythingOfType("string"), mock.AnythingOfType("float64"))
+
+	channelID := model.NewId()
+	userID := model.NewId()
+	connID := model.NewId()
+	originalConnID := model.NewId()
+	prevConnID := model.NewId()
+
+	mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionCreatePost).Return(true).Once()
+
+	// wsEventError must be published back to the failing client.
+	published := make(chan struct{})
+	mockMetrics.On("IncWebSocketEvent", "out", wsEventError).Once()
+	mockAPI.On("PublishWebSocketEvent", wsEventError,
+		map[string]interface{}{"data": "no call ongoing", "connID": connID},
+		&model.WebsocketBroadcast{ConnectionId: connID, ReliableClusterSend: true},
+	).Run(func(_ mock.Arguments) { close(published) }).Return().Once()
+
+	p.WebSocketMessageHasBeenPosted(connID, userID, &model.WebSocketRequest{
+		Action: wsActionPrefix + clientMessageTypeReconnect,
+		Data: map[string]interface{}{
+			"channelID":      channelID,
+			"originalConnID": originalConnID,
+			"prevConnID":     prevConnID,
+		},
+		Session: model.Session{Id: model.NewId()},
+	})
+
+	select {
+	case <-published:
+	case <-time.After(5 * time.Second):
+		t.Fatal("wsEventError was not published after reconnect failure")
+	}
+
+	mockAPI.AssertExpectations(t)
+	mockMetrics.AssertExpectations(t)
+}