Use more reliable workflow mutation check (#4076)

* Make workflow mutation check more reliable

Namespace migration breaks one of the assumption of XDC
XDC assumes there will be no case which a namespace has only one cluster & the cluster is not itself, apparently namespace migration can break this assumption

Author: wxing1292

service/history/historyEngine2_test.go

@@ -149,7 +149,7 @@ func (s *engine2Suite) SetupTest() {
 	s.mockClusterMetadata = s.mockShard.Resource.ClusterMetadata
 
 	s.mockEventsCache = s.mockShard.MockEventsCache
-	s.mockNamespaceCache.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(tests.LocalNamespaceEntry, nil).AnyTimes()
+	s.mockNamespaceCache.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(tests.GlobalNamespaceEntry, nil).AnyTimes()
 	s.mockNamespaceCache.EXPECT().GetNamespaceByID(tests.ParentNamespaceID).Return(tests.GlobalParentNamespaceEntry, nil).AnyTimes()
 	s.mockNamespaceCache.EXPECT().GetNamespace(tests.ChildNamespace).Return(tests.GlobalChildNamespaceEntry, nil).AnyTimes()
 	s.mockEventsCache.EXPECT().PutEvent(gomock.Any(), gomock.Any()).AnyTimes()

service/history/workflow/mutable_state_impl.go

@@ -4239,43 +4239,44 @@ func (ms *MutableStateImpl) eventsToReplicationTask(
 	transactionPolicy TransactionPolicy,
 	events []*historypb.HistoryEvent,
 ) error {
-
-	if transactionPolicy == TransactionPolicyPassive ||
-		!ms.canReplicateEvents() ||
-		len(events) == 0 {
+	switch transactionPolicy {
+	case TransactionPolicyActive:
+		if ms.generateReplicationTask() {
+			return ms.taskGenerator.GenerateHistoryReplicationTasks(events)
+		}
 		return nil
+	case TransactionPolicyPassive:
+		return nil
+	default:
+		panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
 	}
-
-	currentBranchToken, err := ms.GetCurrentBranchToken()
-	if err != nil {
-		return err
-	}
-	return ms.taskGenerator.GenerateHistoryReplicationTasks(
-		currentBranchToken,
-		events,
-	)
 }
 
 func (ms *MutableStateImpl) syncActivityToReplicationTask(
 	now time.Time,
 	transactionPolicy TransactionPolicy,
 ) []tasks.Task {
 
-	if transactionPolicy == TransactionPolicyPassive ||
-		!ms.canReplicateEvents() {
+	switch transactionPolicy {
+	case TransactionPolicyActive:
+		if ms.generateReplicationTask() {
+			return convertSyncActivityInfos(
+				now,
+				definition.NewWorkflowKey(
+					ms.executionInfo.NamespaceId,
+					ms.executionInfo.WorkflowId,
+					ms.executionState.RunId,
+				),
+				ms.pendingActivityInfoIDs,
+				ms.syncActivityTasks,
+			)
+		}
+		return nil
+	case TransactionPolicyPassive:
 		return emptyTasks
+	default:
+		panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
 	}
-
-	return convertSyncActivityInfos(
-		now,
-		definition.NewWorkflowKey(
-			ms.executionInfo.NamespaceId,
-			ms.executionInfo.WorkflowId,
-			ms.executionState.RunId,
-		),
-		ms.pendingActivityInfoIDs,
-		ms.syncActivityTasks,
-	)
 }
 
 func (ms *MutableStateImpl) updatePendingEventIDs(
@@ -4319,10 +4320,6 @@ func (ms *MutableStateImpl) updateWithLastWriteEvent(
 	return nil
 }
 
-func (ms *MutableStateImpl) canReplicateEvents() bool {
-	return ms.namespaceEntry.ReplicationPolicy() == namespace.ReplicationPolicyMultiCluster
-}
-
 // validateNoEventsAfterWorkflowFinish perform check on history event batch
 // NOTE: do not apply this check on every batch, since transient
 // workflow task && workflow finish will be broken (the first batch)
@@ -4478,23 +4475,24 @@ func (ms *MutableStateImpl) startTransactionHandleWorkflowTaskFailover() (bool,
 func (ms *MutableStateImpl) closeTransactionWithPolicyCheck(
 	transactionPolicy TransactionPolicy,
 ) error {
-
-	if transactionPolicy == TransactionPolicyPassive ||
-		!ms.canReplicateEvents() {
+	switch transactionPolicy {
+	case TransactionPolicyActive:
+		// Cannot use ms.namespaceEntry.ActiveClusterName() because currentVersion may be updated during this transaction in
+		// passive cluster. For example: if passive cluster sees conflict and decided to terminate this workflow. The
+		// currentVersion on mutable state would be updated to point to last write version which is current (passive) cluster.
+		activeCluster := ms.clusterMetadata.ClusterNameForFailoverVersion(ms.namespaceEntry.IsGlobalNamespace(), ms.GetCurrentVersion())
+		currentCluster := ms.clusterMetadata.GetCurrentClusterName()
+
+		if activeCluster != currentCluster {
+			namespaceID := ms.GetExecutionInfo().NamespaceId
+			return serviceerror.NewNamespaceNotActive(namespaceID, currentCluster, activeCluster)
+		}
 		return nil
+	case TransactionPolicyPassive:
+		return nil
+	default:
+		panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
 	}
-
-	// Cannot use ms.namespaceEntry.ActiveClusterName() because currentVersion may be updated during this transaction in
-	// passive cluster. For example: if passive cluster sees conflict and decided to terminate this workflow. The
-	// currentVersion on mutable state would be updated to point to last write version which is current (passive) cluster.
-	activeCluster := ms.clusterMetadata.ClusterNameForFailoverVersion(ms.namespaceEntry.IsGlobalNamespace(), ms.GetCurrentVersion())
-	currentCluster := ms.clusterMetadata.GetCurrentClusterName()
-
-	if activeCluster != currentCluster {
-		namespaceID := ms.GetExecutionInfo().NamespaceId
-		return serviceerror.NewNamespaceNotActive(namespaceID, currentCluster, activeCluster)
-	}
-	return nil
 }
 
 func (ms *MutableStateImpl) closeTransactionHandleBufferedEventsLimit(
@@ -4573,17 +4571,24 @@ func (ms *MutableStateImpl) closeTransactionHandleWorkflowReset(
 func (ms *MutableStateImpl) closeTransactionHandleActivityUserTimerTasks(
 	transactionPolicy TransactionPolicy,
 ) error {
-
-	if transactionPolicy == TransactionPolicyPassive ||
-		!ms.IsWorkflowExecutionRunning() {
+	switch transactionPolicy {
+	case TransactionPolicyActive:
+		if !ms.IsWorkflowExecutionRunning() {
+			return nil
+		}
+		if err := ms.taskGenerator.GenerateActivityTimerTasks(); err != nil {
+			return err
+		}
+		return ms.taskGenerator.GenerateUserTimerTasks()
+	case TransactionPolicyPassive:
 		return nil
+	default:
+		panic(fmt.Sprintf("unknown transaction policy: %v", transactionPolicy))
 	}
+}
 
-	if err := ms.taskGenerator.GenerateActivityTimerTasks(); err != nil {
-		return err
-	}
-
-	return ms.taskGenerator.GenerateUserTimerTasks()
+func (ms *MutableStateImpl) generateReplicationTask() bool {
+	return len(ms.namespaceEntry.ClusterNames()) > 1
 }
 
 func (ms *MutableStateImpl) checkMutability(

service/history/workflow/task_generator.go

@@ -99,7 +99,6 @@ type (
 
 		// replication tasks
 		GenerateHistoryReplicationTasks(
-			branchToken []byte,
 			events []*historypb.HistoryEvent,
 		) error
 		GenerateMigrationTasks() (tasks.Task, error)
@@ -569,9 +568,12 @@ func (r *TaskGeneratorImpl) GenerateUserTimerTasks() error {
 }
 
 func (r *TaskGeneratorImpl) GenerateHistoryReplicationTasks(
-	branchToken []byte,
 	events []*historypb.HistoryEvent,
 ) error {
+	if len(events) == 0 {
+		return nil
+	}
+
 	firstEvent := events[0]
 	lastEvent := events[len(events)-1]
 	if firstEvent.GetVersion() != lastEvent.GetVersion() {