From 5433e6eef9e9fb16b623a4a851ec9a21352a2f49 Mon Sep 17 00:00:00 2001 From: "cursor[bot]" <206951365+cursor[bot]@users.noreply.github.com> Date: Fri, 3 Jul 2026 23:06:27 -0400 Subject: [PATCH] [MM-69689] Self-heal orphaned shared channel remotes for deleted remote clusters (#37346) * [MM-69689] Self-heal orphaned shared channel remotes for deleted remote clusters When a SharedChannelRemote row is still live (DeleteAt = 0) but its RemoteCluster has been soft-deleted, the sync loop repeatedly picks up the orphaned row and logs a warning on every tick without ever cleaning it up. Soft-delete the orphaned SharedChannelRemote when the remote cluster lookup returns sql.ErrNoRows, so the row stops recurring in the sync loop. Logs a Warn (not Error) and skips the task. Co-authored-by: mattermost-code * [MM-69689] Expand self-heal test coverage for orphaned shared channel remotes Add subtests covering the DeleteRemote failure branch and an end-to-end doSync run proving a self-healed orphan is drained and not retried, and assert DeleteRemote call counts. Co-authored-by: mattermost-code * [MM-69689] Log already-deleted remote skip at debug instead of warn A soft-deleted RemoteCluster no longer syncing is the expected steady state, so skipping it does not warrant a warning. Downgrade the log for the already-soft-deleted SCR case to debug to avoid noise. --------- Co-authored-by: Cursor Agent Co-authored-by: mattermost-code --- .../services/sharedchannel/sync_send.go | 50 +++++++++- .../services/sharedchannel/sync_send_test.go | 95 +++++++++++++++++-- 2 files changed, 131 insertions(+), 14 deletions(-) diff --git a/server/platform/services/sharedchannel/sync_send.go b/server/platform/services/sharedchannel/sync_send.go index e4261e6e3fa4..f578ff999899 100644 --- a/server/platform/services/sharedchannel/sync_send.go +++ b/server/platform/services/sharedchannel/sync_send.go @@ -436,10 +436,7 @@ func (scs *Service) processTask(task syncTask) error { // nothing to sync, so drop the task rather than retrying and logging // an error on every change for the orphaned reference. if errors.Is(err, sql.ErrNoRows) { - scs.server.Log().Warn("Skipping sync for deleted remote cluster", - mlog.String("channelId", task.channelID), - mlog.String("remoteId", task.remoteID), - ) + scs.selfHealOrphanedSharedChannelRemote(task.channelID, task.remoteID) return nil } return err @@ -469,6 +466,51 @@ func (scs *Service) processTask(task syncTask) error { return nil } +// selfHealOrphanedSharedChannelRemote handles a SharedChannelRemote row that still +// references a RemoteCluster which has been soft-deleted or removed. Left alone, the +// live SCR row (DeleteAt = 0) keeps getting picked up by the sync loop, spamming a log +// entry on every tick. Soft-deleting the orphaned SCR row stops the recurrence without +// requiring manual DB intervention. +func (scs *Service) selfHealOrphanedSharedChannelRemote(channelID, remoteID string) { + scr, err := scs.server.GetStore().SharedChannel().GetRemoteByIds(channelID, remoteID) + if err != nil { + // The SCR row is already gone (or unreadable); nothing left to self-heal. + scs.server.Log().Warn("Skipping sync for deleted remote cluster", + mlog.String("channelId", channelID), + mlog.String("remoteId", remoteID), + mlog.Err(err), + ) + return + } + + if scr.DeleteAt != 0 { + // The SCR row is already soft-deleted, so there is nothing to heal. A deleted + // remote no longer syncing is the expected steady state, so this is logged at + // debug rather than warn to avoid noise. + scs.server.Log().Debug("Skipping sync for deleted remote cluster", + mlog.String("channelId", channelID), + mlog.String("remoteId", remoteID), + ) + return + } + + if _, err := scs.server.GetStore().SharedChannel().DeleteRemote(scr.Id); err != nil { + scs.server.Log().Warn("Failed to self-heal orphaned shared channel remote for deleted remote cluster", + mlog.String("channelId", channelID), + mlog.String("remoteId", remoteID), + mlog.String("sharedChannelRemoteId", scr.Id), + mlog.Err(err), + ) + return + } + + scs.server.Log().Warn("Self-healed orphaned shared channel remote for deleted remote cluster", + mlog.String("channelId", channelID), + mlog.String("remoteId", remoteID), + mlog.String("sharedChannelRemoteId", scr.Id), + ) +} + func (scs *Service) handlePostError(postId string, task syncTask, rc *model.RemoteCluster) { if task.retryMsg != nil && len(task.retryMsg.Posts) == 1 && task.retryMsg.Posts[0].Id == postId { // this was a retry for specific post that failed previously. Try again if within MaxRetries. diff --git a/server/platform/services/sharedchannel/sync_send_test.go b/server/platform/services/sharedchannel/sync_send_test.go index 367dc119d7a2..bbc9e421a28d 100644 --- a/server/platform/services/sharedchannel/sync_send_test.go +++ b/server/platform/services/sharedchannel/sync_send_test.go @@ -11,8 +11,10 @@ import ( "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/v8/channels/store" "github.com/mattermost/mattermost/server/v8/channels/store/storetest/mocks" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) @@ -24,47 +26,120 @@ func newTestService() *Service { } func TestProcessTask_RemoteClusterLookup(t *testing.T) { - newServiceWithRemoteGet := func(t *testing.T, remoteID string, rc *model.RemoteCluster, getErr error) *Service { + // Mirror how the store wraps the not-found case for a soft-deleted/missing remote. + notFound := fmt.Errorf("failed to find RemoteCluster: %w", sql.ErrNoRows) + + newService := func(t *testing.T, remoteID string, rc *model.RemoteCluster, getErr error) (*Service, *mocks.SharedChannelStore) { t.Helper() mockRemoteClusterStore := &mocks.RemoteClusterStore{} mockRemoteClusterStore.On("Get", remoteID, false).Return(rc, getErr) + mockSharedChannelStore := &mocks.SharedChannelStore{} + mockStore := &mocks.Store{} mockStore.On("RemoteCluster").Return(mockRemoteClusterStore) + mockStore.On("SharedChannel").Return(mockSharedChannelStore) mockServer := &MockServerIface{} mockServer.On("GetStore").Return(mockStore) + mockServer.On("GetMetrics").Return(nil) mockServer.On("Log").Return(mlog.CreateConsoleTestLogger(t)) return &Service{ server: mockServer, changeSignal: make(chan struct{}, 1), tasks: make(map[string]syncTask), - } + }, mockSharedChannelStore } - t.Run("deleted or missing remote is skipped without error or retry", func(t *testing.T) { + t.Run("orphaned remote self-heals by soft-deleting the live SCR row", func(t *testing.T) { + channelID := model.NewId() remoteID := model.NewId() - // Mirror how the store wraps the not-found case. - notFound := fmt.Errorf("failed to find RemoteCluster: %w", sql.ErrNoRows) - scs := newServiceWithRemoteGet(t, remoteID, nil, notFound) + scr := &model.SharedChannelRemote{Id: model.NewId(), ChannelId: channelID, RemoteId: remoteID, DeleteAt: 0} - err := scs.processTask(newSyncTask("channel-1", "", remoteID, nil, nil)) + scs, scStore := newService(t, remoteID, nil, notFound) + scStore.On("GetRemoteByIds", channelID, remoteID).Return(scr, nil) + scStore.On("DeleteRemote", scr.Id).Return(true, nil) + + err := scs.processTask(newSyncTask(channelID, "", remoteID, nil, nil)) + + require.NoError(t, err, "an orphaned remote should be skipped, not surfaced as an error") + scStore.AssertNumberOfCalls(t, "DeleteRemote", 1) + scStore.AssertCalled(t, "DeleteRemote", scr.Id) + }) + + t.Run("self-healed orphan is drained from the sync queue and not retried", func(t *testing.T) { + channelID := model.NewId() + remoteID := model.NewId() + scr := &model.SharedChannelRemote{Id: model.NewId(), ChannelId: channelID, RemoteId: remoteID, DeleteAt: 0} + + scs, scStore := newService(t, remoteID, nil, notFound) + scStore.On("GetRemoteByIds", channelID, remoteID).Return(scr, nil) + scStore.On("DeleteRemote", scr.Id).Return(true, nil) + + task := newSyncTask(channelID, "", remoteID, nil, nil) + scs.addTask(task) + + scs.doSync() + + assert.Empty(t, scs.tasks, "the orphaned task must be drained, not re-enqueued for retry") + scStore.AssertNumberOfCalls(t, "DeleteRemote", 1) + }) + + t.Run("already soft-deleted SCR row is not deleted again", func(t *testing.T) { + channelID := model.NewId() + remoteID := model.NewId() + scr := &model.SharedChannelRemote{Id: model.NewId(), ChannelId: channelID, RemoteId: remoteID, DeleteAt: model.GetMillis()} + + scs, scStore := newService(t, remoteID, nil, notFound) + scStore.On("GetRemoteByIds", channelID, remoteID).Return(scr, nil) + + err := scs.processTask(newSyncTask(channelID, "", remoteID, nil, nil)) + + require.NoError(t, err) + scStore.AssertNotCalled(t, "DeleteRemote", mock.Anything) + }) + + t.Run("missing SCR row is skipped without error or delete", func(t *testing.T) { + channelID := model.NewId() + remoteID := model.NewId() + + scs, scStore := newService(t, remoteID, nil, notFound) + scStore.On("GetRemoteByIds", channelID, remoteID).Return(nil, store.NewErrNotFound("SharedChannelRemote", "missing")) + + err := scs.processTask(newSyncTask(channelID, "", remoteID, nil, nil)) + + require.NoError(t, err) + scStore.AssertNotCalled(t, "DeleteRemote", mock.Anything) + }) + + t.Run("failure to soft-delete the orphan is swallowed without retry", func(t *testing.T) { + channelID := model.NewId() + remoteID := model.NewId() + scr := &model.SharedChannelRemote{Id: model.NewId(), ChannelId: channelID, RemoteId: remoteID, DeleteAt: 0} + + scs, scStore := newService(t, remoteID, nil, notFound) + scStore.On("GetRemoteByIds", channelID, remoteID).Return(scr, nil) + scStore.On("DeleteRemote", scr.Id).Return(false, errors.New("db is down")) + + err := scs.processTask(newSyncTask(channelID, "", remoteID, nil, nil)) - require.NoError(t, err, "a deleted remote should be skipped, not surfaced as an error") - assert.Empty(t, scs.tasks, "the task should not be re-enqueued for retry") + require.NoError(t, err, "a self-heal delete failure must not surface as a sync error or trigger a retry") + scStore.AssertCalled(t, "DeleteRemote", scr.Id) }) t.Run("transient error is propagated for retry", func(t *testing.T) { remoteID := model.NewId() dbErr := errors.New("write tcp: connection reset by peer") - scs := newServiceWithRemoteGet(t, remoteID, nil, dbErr) + scs, scStore := newService(t, remoteID, nil, dbErr) err := scs.processTask(newSyncTask("channel-1", "", remoteID, nil, nil)) require.Error(t, err, "a transient lookup error must still be returned so the task retries") assert.ErrorContains(t, err, "connection reset by peer") + scStore.AssertNotCalled(t, "GetRemoteByIds", mock.Anything, mock.Anything) + scStore.AssertNotCalled(t, "DeleteRemote", mock.Anything) }) }