Calls v1: Reconcile RTCD sessions to clean up orphaned DB rows [MM-70185]

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),

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)
+		}
+	}
+}

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)
+	})
+}