From f40dd1271acfeaa82374ae81d400a3ef1791655d Mon Sep 17 00:00:00 2001 From: Bill Gardner Date: Wed, 19 Aug 2026 18:46:04 -0400 Subject: [PATCH] [MM-70185] Reconcile RTCD sessions to clean up orphaned DB rows When the app node that owns a session's RTCD WebSocket connection dies, RTCD can no longer deliver ClientMessageClose for that session. The calls_sessions row stays indefinitely, blocking the call from ending even after all real participants have left. Add a 30s ticker on the plugin side that compares calls_sessions DB rows against GetSessions on RTCD and removes any row RTCD no longer knows about. If all sessions for a call are orphaned, also clean up the call state so the channel is unblocked for new calls. --- server/activate.go | 3 + server/rtcd_reconciler.go | 122 +++++++++++++++ server/rtcd_reconciler_test.go | 264 +++++++++++++++++++++++++++++++++ 3 files changed, 389 insertions(+) create mode 100644 server/rtcd_reconciler.go create mode 100644 server/rtcd_reconciler_test.go diff --git a/server/activate.go b/server/activate.go index 75197ebbf..3c4639830 100644 --- a/server/activate.go +++ b/server/activate.go @@ -167,6 +167,9 @@ func (p *Plugin) OnActivate() (retErr error) { if err := p.cleanUpState(); err != nil { p.LogError("failed to cleanup state", "err", err.Error()) } + + go p.runRTCDSessionReconciler() + p.LogDebug("started RTCD session reconciler") } else { rtcServerConfig := rtc.ServerConfig{ ICEAddressUDP: rtc.ICEAddress(cfg.UDPServerAddress), diff --git a/server/rtcd_reconciler.go b/server/rtcd_reconciler.go new file mode 100644 index 000000000..d7db54354 --- /dev/null +++ b/server/rtcd_reconciler.go @@ -0,0 +1,122 @@ +// Copyright (c) 2020-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package main + +import ( + "net/http" + "strings" + "time" + + "github.com/mattermost/mattermost-plugin-calls/server/db" +) + +const rtcdSessionReconcilerInterval = 30 * time.Second + +func (p *Plugin) runRTCDSessionReconciler() { + ticker := time.NewTicker(rtcdSessionReconcilerInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + p.reconcileRTCDSessions() + case <-p.stopCh: + return + } + } +} + +// reconcileRTCDSessions compares calls_sessions DB rows against what RTCD +// reports for each active call and cleans up any rows RTCD no longer knows +// about. These orphaned rows occur when the app node that owned a session's +// WebSocket connection dies before RTCD can deliver the close event. +// +// If all sessions for a call are orphaned (RTCD has no sessions), the call +// state is also cleaned up — otherwise the call would remain active +// indefinitely, blocking new calls in the channel. +func (p *Plugin) reconcileRTCDSessions() { + calls, err := p.store.GetAllActiveCalls(db.GetCallOpts{FromWriter: true}) + if err != nil { + p.LogError("rtcd reconciler: failed to get active calls", "err", err.Error()) + return + } + + for _, call := range calls { + if call.Props.RTCDHost == "" { + continue + } + + host := p.rtcdManager.getHost(call.Props.RTCDHost) + if host == nil { + // RTCD node is gone entirely; cleanUpState handles this path. + continue + } + + // GetSessions requires RTCD v1.0.0+. + info, err := host.client.GetVersionInfo() + if err != nil { + p.LogDebug("rtcd reconciler: failed to get version info", "err", err.Error(), "callID", call.ID, "rtcdHost", call.Props.RTCDHost) + continue + } + if info.BuildVersion != "" && info.BuildVersion != "master" && !strings.HasPrefix(info.BuildVersion, "dev") { + if err := checkMinVersion("v1.0.0", info.BuildVersion); err != nil { + p.LogDebug("rtcd reconciler: RTCD version does not support GetSessions", "err", err.Error(), "callID", call.ID) + continue + } + } + + cfgs, code, err := host.client.GetSessions(call.ID) + if err != nil || (code != http.StatusOK && code != http.StatusNotFound) { + p.LogDebug("rtcd reconciler: failed to get sessions from RTCD", "err", err, "code", code, "callID", call.ID) + continue + } + + rtcdSessionIDs := make(map[string]struct{}, len(cfgs)) + for _, cfg := range cfgs { + rtcdSessionIDs[cfg.SessionID] = struct{}{} + } + + dbSessions, err := p.store.GetCallSessions(call.ID, db.GetCallSessionOpts{}) + if err != nil { + p.LogError("rtcd reconciler: failed to get DB sessions", "err", err.Error(), "callID", call.ID) + continue + } + + var orphaned int + for sessionID := range dbSessions { + if _, ok := rtcdSessionIDs[sessionID]; !ok { + p.LogInfo("rtcd reconciler: deleting orphaned session", "sessionID", sessionID, "callID", call.ID) + if err := p.store.DeleteCallSession(sessionID); err != nil { + p.LogError("rtcd reconciler: failed to delete orphaned session", "err", err.Error(), "sessionID", sessionID) + } else { + orphaned++ + } + } + } + + // If RTCD has no sessions for this call, the call has ended but the + // plugin never received the close events. Clean up call state now so + // the channel doesn't remain blocked indefinitely. + if len(cfgs) == 0 && orphaned > 0 { + p.LogInfo("rtcd reconciler: all sessions were orphaned, cleaning up call state", "callID", call.ID, "channelID", call.ChannelID) + + state, err := p.lockCallReturnState(call.ChannelID) + if err != nil { + p.LogError("rtcd reconciler: failed to lock call", "err", err.Error(), "callID", call.ID) + continue + } + + // Re-check under lock: another node or path may have raced us. + if state == nil || len(state.sessions) > 0 { + p.unlockCall(call.ChannelID) + continue + } + + if err := p.cleanCallState(&state.Call); err != nil { + p.LogError("rtcd reconciler: failed to clean call state", "err", err.Error(), "callID", call.ID) + } + p.unlockCall(call.ChannelID) + } + } +} diff --git a/server/rtcd_reconciler_test.go b/server/rtcd_reconciler_test.go new file mode 100644 index 000000000..db7fe22a4 --- /dev/null +++ b/server/rtcd_reconciler_test.go @@ -0,0 +1,264 @@ +// Copyright (c) 2020-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package main + +import ( + "testing" + "time" + + "github.com/mattermost/mattermost-plugin-calls/server/cluster" + "github.com/mattermost/mattermost-plugin-calls/server/db" + "github.com/mattermost/mattermost-plugin-calls/server/public" + rtcd "github.com/mattermost/rtcd/service" + "github.com/mattermost/rtcd/service/rtc" + + serverMocks "github.com/mattermost/mattermost-plugin-calls/server/mocks/github.com/mattermost/mattermost-plugin-calls/server/interfaces" + pluginMocks "github.com/mattermost/mattermost-plugin-calls/server/mocks/github.com/mattermost/mattermost/server/public/plugin" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/plugin" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestReconcileRTCDSessions(t *testing.T) { + mockAPI := &pluginMocks.MockAPI{} + mockMetrics := &serverMocks.MockMetrics{} + + p := &Plugin{ + MattermostPlugin: plugin.MattermostPlugin{ + API: mockAPI, + }, + metrics: mockMetrics, + callsClusterLocks: map[string]*cluster.Mutex{}, + sessions: map[string]*session{}, + } + + store, tearDown := NewTestStore(t) + t.Cleanup(tearDown) + p.store = store + + mockMetrics.On("ObserveAppHandlersTime", mock.AnythingOfType("string"), mock.AnythingOfType("float64")).Maybe() + mockMetrics.On("ObserveClusterMutexGrabTime", "mutex_call", mock.AnythingOfType("float64")).Maybe() + mockMetrics.On("ObserveClusterMutexLockedTime", "mutex_call", mock.AnythingOfType("float64")).Maybe() + 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).Maybe() + mockAPI.On("LogError", mock.Anything, mock.Anything, mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything).Maybe() + mockAPI.On("LogInfo", mock.Anything, mock.Anything, mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything).Maybe() + + newCall := func(t *testing.T, channelID, postID, userID string, rtcdHost string) *public.Call { + t.Helper() + call := &public.Call{ + ID: model.NewId(), + CreateAt: time.Now().UnixMilli(), + ChannelID: channelID, + StartAt: time.Now().UnixMilli(), + PostID: postID, + ThreadID: model.NewId(), + OwnerID: userID, + Props: public.CallProps{ + RTCDHost: rtcdHost, + }, + } + err := p.store.CreateCall(call) + require.NoError(t, err) + return call + } + + newSession := func(t *testing.T, callID, sessionID, userID string) { + t.Helper() + err := p.store.CreateCallSession(&public.CallSession{ + ID: sessionID, + CallID: callID, + UserID: userID, + JoinAt: time.Now().UnixMilli(), + }) + require.NoError(t, err) + } + + t.Run("no active calls", func(_ *testing.T) { + p.rtcdManager = &rtcdClientManager{ + ctx: p, + hosts: map[string]*rtcdHost{}, + } + // Should be a no-op without errors. + p.reconcileRTCDSessions() + }) + + t.Run("call without rtcd host skipped", func(t *testing.T) { + defer ResetTestStore(t, p.store) + + channelID := model.NewId() + postID := model.NewId() + userID := model.NewId() + + call := newCall(t, channelID, postID, userID, "") + newSession(t, call.ID, model.NewId(), userID) + + p.rtcdManager = &rtcdClientManager{ + ctx: p, + hosts: map[string]*rtcdHost{}, + } + + p.reconcileRTCDSessions() + + // Session should be untouched. + sessions, err := p.store.GetCallSessions(call.ID, db.GetCallSessionOpts{}) + require.NoError(t, err) + require.Len(t, sessions, 1) + }) + + t.Run("rtcd host not found in manager skipped", func(t *testing.T) { + defer ResetTestStore(t, p.store) + + channelID := model.NewId() + postID := model.NewId() + userID := model.NewId() + + call := newCall(t, channelID, postID, userID, "127.0.0.1") + newSession(t, call.ID, model.NewId(), userID) + + // Manager has no hosts configured. + p.rtcdManager = &rtcdClientManager{ + ctx: p, + hosts: map[string]*rtcdHost{}, + } + + p.reconcileRTCDSessions() + + // Session should be untouched — cleanUpState handles the gone-host path. + sessions, err := p.store.GetCallSessions(call.ID, db.GetCallSessionOpts{}) + require.NoError(t, err) + require.Len(t, sessions, 1) + }) + + t.Run("all sessions live in rtcd, no orphans", func(t *testing.T) { + defer ResetTestStore(t, p.store) + + channelID := model.NewId() + postID := model.NewId() + userID := model.NewId() + sessionID := model.NewId() + + call := newCall(t, channelID, postID, userID, "127.0.0.1") + newSession(t, call.ID, sessionID, userID) + + mockRTCDClient := &serverMocks.MockRTCDClient{} + defer mockRTCDClient.AssertExpectations(t) + + p.rtcdManager = &rtcdClientManager{ + ctx: p, + hosts: map[string]*rtcdHost{ + "127.0.0.1": {client: mockRTCDClient}, + }, + } + + mockRTCDClient.On("GetVersionInfo").Return(rtcd.VersionInfo{}, nil).Once() + mockRTCDClient.On("GetSessions", call.ID).Return([]rtc.SessionConfig{ + {SessionID: sessionID}, + }, 200, nil).Once() + + p.reconcileRTCDSessions() + + // Session and call should be untouched. + sessions, err := p.store.GetCallSessions(call.ID, db.GetCallSessionOpts{}) + require.NoError(t, err) + require.Len(t, sessions, 1) + + calls, err := p.store.GetAllActiveCalls(db.GetCallOpts{}) + require.NoError(t, err) + require.Len(t, calls, 1) + }) + + t.Run("one orphaned session among live ones", func(t *testing.T) { + defer ResetTestStore(t, p.store) + + channelID := model.NewId() + postID := model.NewId() + userID := model.NewId() + liveSessionID := model.NewId() + orphanedSessionID := model.NewId() + + call := newCall(t, channelID, postID, userID, "127.0.0.1") + newSession(t, call.ID, liveSessionID, userID) + newSession(t, call.ID, orphanedSessionID, model.NewId()) + + mockRTCDClient := &serverMocks.MockRTCDClient{} + defer mockRTCDClient.AssertExpectations(t) + + p.rtcdManager = &rtcdClientManager{ + ctx: p, + hosts: map[string]*rtcdHost{ + "127.0.0.1": {client: mockRTCDClient}, + }, + } + + // RTCD only knows about the live session. + mockRTCDClient.On("GetVersionInfo").Return(rtcd.VersionInfo{}, nil).Once() + mockRTCDClient.On("GetSessions", call.ID).Return([]rtc.SessionConfig{ + {SessionID: liveSessionID}, + }, 200, nil).Once() + + p.reconcileRTCDSessions() + + // Orphaned session deleted, live session kept, call still active. + sessions, err := p.store.GetCallSessions(call.ID, db.GetCallSessionOpts{}) + require.NoError(t, err) + require.Len(t, sessions, 1) + require.NotNil(t, sessions[liveSessionID]) + + calls, err := p.store.GetAllActiveCalls(db.GetCallOpts{}) + require.NoError(t, err) + require.Len(t, calls, 1) + }) + + t.Run("all sessions orphaned, call state cleaned up", func(t *testing.T) { + defer ResetTestStore(t, p.store) + + channelID := model.NewId() + postID := model.NewId() + userID := model.NewId() + sessionID := model.NewId() + + call := newCall(t, channelID, postID, userID, "127.0.0.1") + createPost(t, store, postID, userID, channelID) + newSession(t, call.ID, sessionID, userID) + + mockRTCDClient := &serverMocks.MockRTCDClient{} + defer mockRTCDClient.AssertExpectations(t) + + p.rtcdManager = &rtcdClientManager{ + ctx: p, + hosts: map[string]*rtcdHost{ + "127.0.0.1": {client: mockRTCDClient}, + }, + } + + // RTCD has no sessions for this call — they've all ended. + mockRTCDClient.On("GetVersionInfo").Return(rtcd.VersionInfo{}, nil).Once() + mockRTCDClient.On("GetSessions", call.ID).Return(nil, 404, nil).Once() + + mockAPI.On("KVSetWithOptions", mock.Anything, mock.Anything, mock.Anything).Return(true, nil).Once() + mockAPI.On("KVDelete", "mutex_call_"+channelID).Return(nil).Once() + mockAPI.On("UpdatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: postID}, nil).Once() + mockAPI.On("GetConfig").Return(&model.Config{}, nil).Once() + + p.reconcileRTCDSessions() + + // Both the session and the call should be cleaned up. + sessions, err := p.store.GetCallSessions(call.ID, db.GetCallSessionOpts{}) + require.NoError(t, err) + require.Empty(t, sessions) + + calls, err := p.store.GetAllActiveCalls(db.GetCallOpts{}) + require.NoError(t, err) + require.Empty(t, calls) + }) +}