Backport #12323 to release/v1.30.x

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,

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:

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)

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

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

service/history/replication/executable_task.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"sync/atomic"
 	"time"
 
@@ -762,13 +763,31 @@ 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 {
 				return "", false, err
 			}
 		}
 	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 {