diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index d9784f6777d..fb6fa781446 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -2501,6 +2501,11 @@ should be enabled for non continuedAsNew workflow UpdateWithNew case.`, false, `ReplicationMultipleBatches is the flag to enable replication of multiple history event batches`, ) + ValidateReplicationTaskSourceCluster = NewGlobalBoolSetting( + "history.validateReplicationTaskSourceCluster", + true, + `ValidateReplicationTaskSourceCluster controls whether inbound workflow replication tasks are accepted only from clusters in the namespace cluster list.`, + ) HistoryTaskDLQEnabled = NewGlobalBoolSetting( "history.TaskDLQEnabled", true, diff --git a/common/namespace/nsreplication/replication_task_executor.go b/common/namespace/nsreplication/replication_task_executor.go index e643c8c8fa8..da63eae4f27 100644 --- a/common/namespace/nsreplication/replication_task_executor.go +++ b/common/namespace/nsreplication/replication_task_executor.go @@ -113,6 +113,15 @@ func (h *taskExecutorImpl) shouldProcessTask(ctx context.Context, task *replicat tag.NewStringTag("Task Namespace Info Id", task.Info.GetId())) return false, ErrNameUUIDCollision } + if !resp.IsGlobalNamespace { + h.logger.Warn( + "Namespace replication task skipped because namespace is local", + tag.WorkflowNamespaceID(resp.Namespace.Info.Id), + tag.WorkflowNamespace(resp.Namespace.Info.Name), + tag.NewStringTag("namespace-operation", task.GetNamespaceOperation().String()), + ) + return false, nil + } return true, nil case *serviceerror.NamespaceNotFound: diff --git a/common/namespace/nsreplication/replication_task_executor_test.go b/common/namespace/nsreplication/replication_task_executor_test.go index ac10bb7688a..df4c3f39419 100644 --- a/common/namespace/nsreplication/replication_task_executor_test.go +++ b/common/namespace/nsreplication/replication_task_executor_test.go @@ -137,7 +137,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_RegisterNamespaceTas Info: &persistencespb.NamespaceInfo{ Id: nsID, }, - }}, nil + }, IsGlobalNamespace: true}, nil }).Times(2) s.mockMetadataMgr.EXPECT().CreateNamespace(gomock.Any(), gomock.Any()).Return(nil, errors.New("test")) err = s.namespaceReplicator.Execute(context.Background(), task) @@ -282,7 +282,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_RegisterNamespaceTas Info: &persistencespb.NamespaceInfo{ Id: id, }, - }}, nil).Times(2) + }, IsGlobalNamespace: true}, nil).Times(2) s.mockMetadataMgr.EXPECT().GetNamespace(gomock.Any(), &persistence.GetNamespaceRequest{ ID: id, }).Return(&persistence.GetNamespaceResponse{Namespace: &persistencespb.NamespaceDetail{ @@ -295,6 +295,37 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_RegisterNamespaceTas s.Nil(err) } +func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateLocalNamespaceTask_NotAdmitted() { + task := &replicationspb.NamespaceTaskAttributes{ + Id: uuid.NewString(), + NamespaceOperation: enumsspb.NAMESPACE_OPERATION_UPDATE, + Info: &namespacepb.NamespaceInfo{ + Name: uuid.NewString(), + State: enumspb.NAMESPACE_STATE_REGISTERED, + }, + Config: &namespacepb.NamespaceConfig{}, + ReplicationConfig: &replicationpb.NamespaceReplicationConfig{ + Clusters: []*replicationpb.ClusterReplicationConfig{ + {ClusterName: s.namespaceReplicator.currentCluster}, + }, + }, + ConfigVersion: 1, + FailoverVersion: 1, + } + s.mockMetadataMgr.EXPECT().GetNamespace(gomock.Any(), &persistence.GetNamespaceRequest{ + Name: task.GetInfo().GetName(), + }).Return(&persistence.GetNamespaceResponse{ + Namespace: &persistencespb.NamespaceDetail{ + Info: &persistencespb.NamespaceInfo{Id: task.GetId()}, + }, + IsGlobalNamespace: false, + }, nil) + s.mockMetadataMgr.EXPECT().UpdateNamespace(gomock.Any(), gomock.Any()).Times(0) + + err := s.namespaceReplicator.Execute(context.Background(), task) + s.Require().NoError(err) +} + func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_NamespaceNotExist() { operation := enumsspb.NAMESPACE_OPERATION_UPDATE id := uuid.NewString() @@ -446,7 +477,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ Id: id, }, ReplicationConfig: &persistencespb.NamespaceReplicationConfig{}, - }}, nil).Times(2) + }, IsGlobalNamespace: true}, nil).Times(2) s.mockMetadataMgr.EXPECT().GetMetadata(gomock.Any()).Return(&persistence.GetMetadataResponse{ NotificationVersion: updateFailoverVersion, }, nil).Times(1) @@ -476,7 +507,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ FailoverNotificationVersion: updateFailoverVersion, FailoverVersion: updateFailoverVersion, }, - IsGlobalNamespace: false, + IsGlobalNamespace: true, NotificationVersion: updateFailoverVersion, }) err := s.namespaceReplicator.Execute(context.Background(), updateTask) @@ -542,7 +573,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ }, ReplicationConfig: &persistencespb.NamespaceReplicationConfig{}, FailoverVersion: updateFailoverVersion + 1, - }}, nil).Times(2) + }, IsGlobalNamespace: true}, nil).Times(2) s.mockMetadataMgr.EXPECT().GetMetadata(gomock.Any()).Return(&persistence.GetMetadataResponse{ NotificationVersion: updateFailoverVersion, }, nil).Times(1) @@ -570,7 +601,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ FailoverNotificationVersion: 0, FailoverVersion: updateFailoverVersion + 1, }, - IsGlobalNamespace: false, + IsGlobalNamespace: true, NotificationVersion: updateFailoverVersion, }) err := s.namespaceReplicator.Execute(context.Background(), updateTask) @@ -636,7 +667,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ }, ReplicationConfig: &persistencespb.NamespaceReplicationConfig{}, ConfigVersion: updateConfigVersion + 1, - }}, nil).Times(2) + }, IsGlobalNamespace: true}, nil).Times(2) s.mockMetadataMgr.EXPECT().GetMetadata(gomock.Any()).Return(&persistence.GetMetadataResponse{ NotificationVersion: updateFailoverVersion, }, nil).Times(1) @@ -652,7 +683,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ FailoverNotificationVersion: updateFailoverVersion, FailoverVersion: updateFailoverVersion, }, - IsGlobalNamespace: false, + IsGlobalNamespace: true, NotificationVersion: updateFailoverVersion, }) err := s.namespaceReplicator.Execute(context.Background(), updateTask) @@ -719,7 +750,7 @@ func (s *namespaceReplicationTaskExecutorSuite) TestExecute_UpdateNamespaceTask_ ReplicationConfig: &persistencespb.NamespaceReplicationConfig{}, ConfigVersion: updateConfigVersion + 1, FailoverVersion: updateFailoverVersion + 1, - }}, nil).Times(2) + }, IsGlobalNamespace: true}, nil).Times(2) s.mockMetadataMgr.EXPECT().GetMetadata(gomock.Any()).Return(&persistence.GetMetadataResponse{ NotificationVersion: updateFailoverVersion, }, nil).Times(1) diff --git a/service/history/configs/config.go b/service/history/configs/config.go index fcf5f501c3a..3b3f0eea6ae 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -274,6 +274,7 @@ type Config struct { ReplicationEnableDLQMetrics dynamicconfig.BoolPropertyFn ReplicationEnableUpdateWithNewTaskMerge dynamicconfig.BoolPropertyFn ReplicationMultipleBatches dynamicconfig.BoolPropertyFn + ValidateReplicationTaskSourceCluster dynamicconfig.BoolPropertyFn ReplicationStreamSenderErrorRetryWait dynamicconfig.DurationPropertyFn ReplicationStreamSenderErrorRetryBackoffCoefficient dynamicconfig.FloatPropertyFn ReplicationStreamSenderErrorRetryMaxInterval dynamicconfig.DurationPropertyFn @@ -581,6 +582,7 @@ func NewConfig( ReplicationTaskProcessorShardQPS: dynamicconfig.ReplicationTaskProcessorShardQPS.Get(dc), ReplicationEnableDLQMetrics: dynamicconfig.ReplicationEnableDLQMetrics.Get(dc), ReplicationEnableUpdateWithNewTaskMerge: dynamicconfig.ReplicationEnableUpdateWithNewTaskMerge.Get(dc), + ValidateReplicationTaskSourceCluster: dynamicconfig.ValidateReplicationTaskSourceCluster.Get(dc), ReplicationStreamSyncStatusDuration: dynamicconfig.ReplicationStreamSyncStatusDuration.Get(dc), ReplicationProcessorSchedulerQueueSize: dynamicconfig.ReplicationProcessorSchedulerQueueSize.Get(dc), ReplicationProcessorSchedulerWorkerCount: dynamicconfig.ReplicationProcessorSchedulerWorkerCount.Subscribe(dc), diff --git a/service/history/replication/executable_history_task_test.go b/service/history/replication/executable_history_task_test.go index 9d10e924a58..616c035a025 100644 --- a/service/history/replication/executable_history_task_test.go +++ b/service/history/replication/executable_history_task_test.go @@ -14,6 +14,7 @@ import ( "go.temporal.io/api/serviceerror" enumsspb "go.temporal.io/server/api/enums/v1" historyspb "go.temporal.io/server/api/history/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" replicationspb "go.temporal.io/server/api/replication/v1" workflowspb "go.temporal.io/server/api/workflow/v1" "go.temporal.io/server/client" @@ -238,6 +239,35 @@ func (s *executableHistoryTaskSuite) TestExecute_Skip_Namespace() { s.NoError(err) } +func (s *executableHistoryTaskSuite) TestExecute_LocalNamespace_SkipsTask() { + s.processToolBox.Config.ValidateReplicationTaskSourceCluster = dynamicconfig.GetBoolPropertyFn(true) + namespaceEntry := namespace.NewNamespaceForTest( + &persistencespb.NamespaceInfo{ + Id: s.replicationTask.NamespaceId, + Name: "local-namespace", + }, + &persistencespb.NamespaceConfig{}, + false, + &persistencespb.NamespaceReplicationConfig{ + ActiveClusterName: cluster.TestCurrentClusterName, + Clusters: []string{cluster.TestCurrentClusterName}, + }, + cluster.TestCurrentClusterInitialFailoverVersion, + ) + s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(s.replicationTask.NamespaceId)).Return(namespaceEntry, nil) + + task := NewExecutableHistoryTask( + s.processToolBox, + s.taskID, + time.Unix(0, rand.Int63()), + s.replicationTask, + cluster.TestAlternativeClusterName, + s.sourceShardKey, + &replicationspb.ReplicationTask{Priority: enumsspb.TASK_PRIORITY_HIGH}, + ) + s.NoError(task.Execute()) +} + func (s *executableHistoryTaskSuite) TestExecute_Err() { s.executableTask.EXPECT().TerminalState().Return(false) err := errors.New("OwO") diff --git a/service/history/replication/executable_task.go b/service/history/replication/executable_task.go index 15deb5d0e76..242daee0b42 100644 --- a/service/history/replication/executable_task.go +++ b/service/history/replication/executable_task.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "sync/atomic" "time" @@ -762,6 +763,16 @@ func (e *ExecutableTaskImpl) GetNamespaceInfo( namespaceEntry, err := e.NamespaceCache.GetNamespaceByID(namespace.ID(namespaceID)) switch err.(type) { case nil: + if e.Config.ValidateReplicationTaskSourceCluster() && + !slices.Contains(namespaceEntry.ClusterNames(businessID), e.sourceClusterName) { + e.Logger.Warn( + "Replication task skipped because source cluster is not in namespace cluster list", + tag.WorkflowNamespaceID(namespaceID), + tag.SourceCluster(e.sourceClusterName), + ) + e.namespace.Store(namespaceEntry.Name()) + return namespaceEntry.Name().String(), false, nil + } if e.replicationTask.VersionedTransition != nil && e.replicationTask.VersionedTransition.NamespaceFailoverVersion > namespaceEntry.FailoverVersion() { _, err = e.ProcessToolBox.EagerNamespaceRefresher.SyncNamespaceFromSourceCluster(ctx, namespace.ID(namespaceID), e.sourceClusterName) if err != nil { @@ -769,6 +780,14 @@ func (e *ExecutableTaskImpl) GetNamespaceInfo( } } case *serviceerror.NamespaceNotFound: + if e.Config.ValidateReplicationTaskSourceCluster() { + e.Logger.Warn( + "Replication task skipped because namespace is not found", + tag.WorkflowNamespaceID(namespaceID), + tag.SourceCluster(e.sourceClusterName), + ) + return "", false, nil + } _, err = e.ProcessToolBox.EagerNamespaceRefresher.SyncNamespaceFromSourceCluster(ctx, namespace.ID(namespaceID), e.sourceClusterName) if err != nil { e.Logger.Info("Failed to SyncNamespaceFromSourceCluster", tag.Error(err)) @@ -790,15 +809,16 @@ func (e *ExecutableTaskImpl) GetNamespaceInfo( if namespaceEntry.State() == enumspb.NAMESPACE_STATE_DELETED { return namespaceEntry.Name().String(), false, nil } - shouldProcessTask := false -FilterLoop: - for _, targetCluster := range namespaceEntry.ClusterNames(businessID) { - if e.ClusterMetadata.GetCurrentClusterName() == targetCluster { - shouldProcessTask = true - break FilterLoop - } + clusterNames := namespaceEntry.ClusterNames(businessID) + if e.Config.ValidateReplicationTaskSourceCluster() && !slices.Contains(clusterNames, e.sourceClusterName) { + e.Logger.Warn( + "Replication task skipped because source cluster is not in refreshed namespace cluster list", + tag.WorkflowNamespaceID(namespaceID), + tag.SourceCluster(e.sourceClusterName), + ) + return namespaceEntry.Name().String(), false, nil } - return namespaceEntry.Name().String(), shouldProcessTask, nil + return namespaceEntry.Name().String(), slices.Contains(clusterNames, e.ClusterMetadata.GetCurrentClusterName()), nil } func (e *ExecutableTaskImpl) MarkPoisonPill() error { diff --git a/service/history/replication/executable_task_test.go b/service/history/replication/executable_task_test.go index 44977e79dc1..923d4722e8a 100644 --- a/service/history/replication/executable_task_test.go +++ b/service/history/replication/executable_task_test.go @@ -26,6 +26,7 @@ import ( "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/collection" "go.temporal.io/server/common/definition" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" @@ -815,7 +816,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_Process() { }, }, } - namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail)) + namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail), namespace.WithGlobalFlag(true)) s.NoError(err) s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntry, nil).AnyTimes() @@ -842,7 +843,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_Skip() { }, }, } - namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail)) + namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail), namespace.WithGlobalFlag(true)) s.NoError(err) s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntry, nil).AnyTimes() @@ -852,6 +853,56 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_Skip() { s.False(toProcess) } +func (s *executableTaskSuite) TestGetNamespaceInfo_LocalNamespaceSkipsRefreshAndProcessing() { + namespaceID := uuid.NewString() + namespaceName := uuid.NewString() + detail := &persistencespb.NamespaceDetail{ + Info: &persistencespb.NamespaceInfo{ + Id: namespaceID, + Name: namespaceName, + }, + Config: &persistencespb.NamespaceConfig{}, + ReplicationConfig: &persistencespb.NamespaceReplicationConfig{ + ActiveClusterName: cluster.TestCurrentClusterName, + Clusters: []string{cluster.TestCurrentClusterName}, + }, + } + namespaceEntry, err := namespace.FromPersistentState( + detail, + namespace.NewDefaultReplicationResolverFactory()(detail), + namespace.WithGlobalFlag(false), + ) + s.NoError(err) + s.task.replicationTask.VersionedTransition = &persistencespb.VersionedTransition{ + NamespaceFailoverVersion: namespaceEntry.FailoverVersion() + 1, + } + s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntry, nil) + + name, toProcess, err := s.task.GetNamespaceInfo(context.Background(), namespaceID, "test-workflow-id") + s.NoError(err) + s.Equal(namespaceName, name) + s.False(toProcess) +} + +func (s *executableTaskSuite) TestGetNamespaceInfo_SourceClusterNotInNamespaceSkipsRefreshAndProcessing() { + namespaceID := uuid.NewString() + namespaceName := uuid.NewString() + detail := &persistencespb.NamespaceDetail{ + Info: &persistencespb.NamespaceInfo{Id: namespaceID, Name: namespaceName}, + Config: &persistencespb.NamespaceConfig{}, + ReplicationConfig: &persistencespb.NamespaceReplicationConfig{ActiveClusterName: cluster.TestCurrentClusterName, Clusters: []string{cluster.TestCurrentClusterName}}, + } + namespaceEntry, err := namespace.FromPersistentState(detail, namespace.NewDefaultReplicationResolverFactory()(detail), namespace.WithGlobalFlag(true)) + s.NoError(err) + s.task.replicationTask.VersionedTransition = &persistencespb.VersionedTransition{NamespaceFailoverVersion: namespaceEntry.FailoverVersion() + 1} + s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntry, nil) + + name, toProcess, err := s.task.GetNamespaceInfo(context.Background(), namespaceID, "test-workflow-id") + s.NoError(err) + s.Equal(namespaceName, name) + s.False(toProcess) +} + func (s *executableTaskSuite) TestGetNamespaceInfo_Deleted() { namespaceID := uuid.NewString() namespaceName := uuid.NewString() @@ -871,7 +922,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_Deleted() { }, }, } - namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail)) + namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail), namespace.WithGlobalFlag(true)) s.NoError(err) s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntry, nil).AnyTimes() @@ -890,6 +941,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_Error() { } func (s *executableTaskSuite) TestGetNamespaceInfo_NotFoundOnCurrentCluster_SyncFromRemoteSuccess() { + s.config.ValidateReplicationTaskSourceCluster = dynamicconfig.GetBoolPropertyFn(false) namespaceID := uuid.NewString() namespaceName := uuid.NewString() factory := namespace.NewDefaultReplicationResolverFactory() @@ -907,7 +959,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_NotFoundOnCurrentCluster_Sync }, }, } - namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail)) + namespaceEntry, err := namespace.FromPersistentState(detail, factory(detail), namespace.WithGlobalFlag(true)) s.NoError(err) // enable feature flag @@ -922,6 +974,16 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_NotFoundOnCurrentCluster_Sync s.True(toProcess) } +func (s *executableTaskSuite) TestGetNamespaceInfo_NamespaceNotFoundSkipsRefreshAndProcessing() { + namespaceID := uuid.NewString() + s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(nil, serviceerror.NewNamespaceNotFound("namespace not found")) + + name, toProcess, err := s.task.GetNamespaceInfo(context.Background(), namespaceID, "test-workflow-id") + s.NoError(err) + s.Empty(name) + s.False(toProcess) +} + func (s *executableTaskSuite) TestGetNamespaceInfo_NamespaceFailoverNotSync_SyncFromRemoteSuccess() { namespaceID := uuid.NewString() namespaceName := uuid.NewString() @@ -966,7 +1028,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_NamespaceFailoverNotSync_Sync }, FailoverVersion: 10, } - namespaceEntryOld, err := namespace.FromPersistentState(detailOld, factory(detailOld)) + namespaceEntryOld, err := namespace.FromPersistentState(detailOld, factory(detailOld), namespace.WithGlobalFlag(true)) s.NoError(err) detailNew := &persistencespb.NamespaceDetail{ Info: &persistencespb.NamespaceInfo{ @@ -983,7 +1045,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_NamespaceFailoverNotSync_Sync }, FailoverVersion: 100, } - namespaceEntryNew, err := namespace.FromPersistentState(detailNew, factory(detailNew)) + namespaceEntryNew, err := namespace.FromPersistentState(detailNew, factory(detailNew), namespace.WithGlobalFlag(true)) s.NoError(err) s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntryOld, nil).Times(1) @@ -1041,7 +1103,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_NamespaceFailoverBehind_Still }, FailoverVersion: 10, } - namespaceEntryOld, err := namespace.FromPersistentState(detailOld, factory(detailOld)) + namespaceEntryOld, err := namespace.FromPersistentState(detailOld, factory(detailOld), namespace.WithGlobalFlag(true)) s.NoError(err) s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(namespaceEntryOld, nil).Times(1) @@ -1056,6 +1118,7 @@ func (s *executableTaskSuite) TestGetNamespaceInfo_NamespaceFailoverBehind_Still } func (s *executableTaskSuite) TestGetNamespaceInfo_NotFoundOnCurrentCluster_SyncFromRemoteFailed() { + s.config.ValidateReplicationTaskSourceCluster = dynamicconfig.GetBoolPropertyFn(false) namespaceID := uuid.NewString() s.namespaceCache.EXPECT().GetNamespaceByID(namespace.ID(namespaceID)).Return(nil, serviceerror.NewNamespaceNotFound("namespace not found")).AnyTimes() diff --git a/service/history/replication/task_executor.go b/service/history/replication/task_executor.go index 87634120afd..7b4061574e3 100644 --- a/service/history/replication/task_executor.go +++ b/service/history/replication/task_executor.go @@ -5,6 +5,7 @@ package replication import ( "context" "errors" + "slices" "time" commonpb "go.temporal.io/api/common/v1" @@ -51,6 +52,7 @@ type ( workflowCache wcache.Cache metricsHandler metrics.Handler logger log.Logger + validateSource func() bool } ) @@ -73,6 +75,7 @@ func NewTaskExecutor( workflowCache: workflowCache, metricsHandler: shardContext.GetMetricsHandler(), logger: shardContext.GetLogger(), + validateSource: shardContext.GetConfig().ValidateReplicationTaskSourceCluster, } } @@ -377,16 +380,17 @@ func (e *taskExecutorImpl) filterTask( } return false, err } - - shouldProcessTask := false -FilterLoop: - for _, targetCluster := range namespaceEntry.ClusterNames(workflowID) { - if e.currentCluster == targetCluster { - shouldProcessTask = true - break FilterLoop - } + clusterNames := namespaceEntry.ClusterNames(workflowID) + if e.validateSource() && !slices.Contains(clusterNames, e.remoteCluster) { + e.logger.Warn( + "Replication task skipped because source cluster is not in namespace cluster list", + tag.WorkflowNamespaceID(namespaceID.String()), + tag.WorkflowID(workflowID), + tag.SourceCluster(e.remoteCluster), + ) + return false, nil } - return shouldProcessTask, nil + return slices.Contains(clusterNames, e.currentCluster), nil } func (e *taskExecutorImpl) cleanupWorkflowExecution(ctx context.Context, namespaceID string, workflowID string, runID string) (retErr error) { diff --git a/service/history/replication/task_executor_test.go b/service/history/replication/task_executor_test.go index 5ff5668d3d3..5a2586e6eb0 100644 --- a/service/history/replication/task_executor_test.go +++ b/service/history/replication/task_executor_test.go @@ -20,6 +20,7 @@ import ( replicationspb "go.temporal.io/server/api/replication/v1" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/collection" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/persistence" "go.temporal.io/server/common/resourcetest" @@ -141,6 +142,37 @@ func (s *taskExecutorSuite) TestFilterTask_NotApply() { s.False(ok) } +func (s *taskExecutorSuite) TestFilterTask_LocalNamespaceNotApply() { + namespaceID := namespace.ID(uuid.NewString()) + s.mockNamespaceCache.EXPECT(). + GetNamespaceByID(namespaceID). + Return(namespace.NewLocalNamespaceForTest(nil, nil, cluster.TestCurrentClusterName), nil) + ok, err := s.replicationTaskExecutor.filterTask(namespaceID, "test-workflow-id", false) + s.NoError(err) + s.False(ok) +} + +func (s *taskExecutorSuite) TestFilterTask_SourceClusterNotApply() { + namespaceID := namespace.ID(uuid.NewString()) + s.mockNamespaceCache.EXPECT().GetNamespaceByID(namespaceID).Return(namespace.NewGlobalNamespaceForTest( + nil, nil, &persistencespb.NamespaceReplicationConfig{Clusters: []string{cluster.TestCurrentClusterName}}, 0, + ), nil) + ok, err := s.replicationTaskExecutor.filterTask(namespaceID, "test-workflow-id", false) + s.NoError(err) + s.False(ok) +} + +func (s *taskExecutorSuite) TestFilterTask_SourceClusterValidationDisabled() { + s.replicationTaskExecutor.validateSource = dynamicconfig.GetBoolPropertyFn(false) + namespaceID := namespace.ID(uuid.NewString()) + s.mockNamespaceCache.EXPECT().GetNamespaceByID(namespaceID).Return(namespace.NewGlobalNamespaceForTest( + nil, nil, &persistencespb.NamespaceReplicationConfig{Clusters: []string{cluster.TestCurrentClusterName}}, 0, + ), nil) + ok, err := s.replicationTaskExecutor.filterTask(namespaceID, "test-workflow-id", false) + s.NoError(err) + s.True(ok) +} + func (s *taskExecutorSuite) TestFilterTask_Error() { namespaceID := namespace.ID(uuid.NewString()) s.mockNamespaceCache.EXPECT(). diff --git a/service/worker/replicator/replication_message_processor.go b/service/worker/replicator/replication_message_processor.go index dc621c1611d..674b6163c19 100644 --- a/service/worker/replicator/replication_message_processor.go +++ b/service/worker/replicator/replication_message_processor.go @@ -244,9 +244,18 @@ func (p *replicationMessageProcessor) handleTaskQueueUserDataReplicationTask( ctx context.Context, attrs *replicationspb.TaskQueueUserDataAttributes, ) error { - _, err := p.namespaceRegistry.GetNamespaceByID(namespace.ID(attrs.GetNamespaceId())) + namespaceEntry, err := p.namespaceRegistry.GetNamespaceByID(namespace.ID(attrs.GetNamespaceId())) switch err.(type) { case nil: + if !namespaceEntry.IsGlobalNamespace() { + p.logger.Warn( + "Task queue user data replication task skipped because namespace is local", + tag.WorkflowNamespaceID(attrs.GetNamespaceId()), + tag.WorkflowTaskQueueName(attrs.GetTaskQueueName()), + tag.SourceCluster(p.sourceCluster), + ) + return nil + } case *serviceerror.NamespaceNotFound: // The namespace in the request isn't registered on this cluster, drop the replication task. // This is okay and enables using the cluster-global replication queue to replicate different namespaces to diff --git a/service/worker/replicator/replication_message_processor_test.go b/service/worker/replicator/replication_message_processor_test.go new file mode 100644 index 00000000000..d54db616dcf --- /dev/null +++ b/service/worker/replicator/replication_message_processor_test.go @@ -0,0 +1,37 @@ +package replicator + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/server/api/matchingservicemock/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + replicationspb "go.temporal.io/server/api/replication/v1" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/namespace" + "go.uber.org/mock/gomock" +) + +func TestHandleTaskQueueUserDataReplicationTask_LocalNamespace(t *testing.T) { + controller := gomock.NewController(t) + registry := namespace.NewMockRegistry(controller) + matchingClient := matchingservicemock.NewMockMatchingServiceClient(controller) + p := &replicationMessageProcessor{ + namespaceRegistry: registry, + matchingClient: matchingClient, + logger: log.NewNoopLogger(), + sourceCluster: "source-cluster", + } + attrs := &replicationspb.TaskQueueUserDataAttributes{NamespaceId: "namespace-id"} + registry.EXPECT().GetNamespaceByID(namespace.ID("namespace-id")).Return( + namespace.NewLocalNamespaceForTest( + &persistencespb.NamespaceInfo{Id: "namespace-id", Name: "payments"}, + nil, + "current-cluster", + ), + nil, + ) + + require.NoError(t, p.handleTaskQueueUserDataReplicationTask(context.Background(), attrs)) +}