From cddd0ac6d0c36cc350986ac5564e821828cedcd3 Mon Sep 17 00:00:00 2001 From: Bill Gardner Date: Wed, 19 Aug 2026 14:27:50 -0400 Subject: [PATCH] [MM-70184] Fix silent handleReconnect failures leaving orphaned sessions When handleReconnect fails, the client now receives wsEventError so it can fall back to a clean rejoin instead of silently retrying until its 30s reconnect budget expires. Also switches getCallState to read from the writer DB (fromWriter=true), matching MM-69201's fix for the same replica-lag class of failure that caused "no call ongoing" / "session not found" errors during pod rolls. --- server/websocket.go | 6 +- server/websocket_test.go | 201 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 206 insertions(+), 1 deletion(-) diff --git a/server/websocket.go b/server/websocket.go index b99b48ceb..2271d0744 100644 --- a/server/websocket.go +++ b/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 diff --git a/server/websocket_test.go b/server/websocket_test.go index 8645e991f..82fd63960 100644 --- a/server/websocket_test.go +++ b/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) +}