[pull] master from mattermost:master

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.

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