fix(active-active): Fix auto-forwarding for active-active domains (#7356)

<!-- Describe what has changed in this PR -->
**What changed?**
- Update active cluster manager with a new method to get current
workflow state and active cluster selection policy
- Fix auto-forwarding for active-active domains
- Add new test case to simulation tests

<!-- Tell your future self why have you made these changes -->
**Why?**
Bug fix

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
unit tests && simulation tests

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

Author: Shaddoll

.github/workflows/replication-simulation.yml

@@ -15,10 +15,10 @@ jobs:
         scenario:
           - activeactive
           - activeactive_child
+          - activeactive_invalid_cluster_attribute
           - activeactive_same_wfid
           - activeactive_same_wfid_signalwithstart
-          # TODO(active-active): Re-enable this scenario once we have fixed the auto-forwarding issue
-          # - activeactive_same_wfid_signalwithstart_delayed
+          - activeactive_same_wfid_signalwithstart_delayed
           - activeactive_cron
           - activeactive_regional_failover
           - activeactive_regional_failover_start_same_wfid

common/activecluster/manager.go

@@ -40,12 +40,13 @@ import (
 type DomainIDToDomainFn func(id string) (*cache.DomainCacheEntry, error)
 
 const (
-	LookupNewWorkflowOpName                          = "LookupNewWorkflow"
-	LookupWorkflowOpName                             = "LookupWorkflow"
-	GetActiveClusterInfoByClusterAttributeOpName     = "GetActiveClusterInfoByClusterAttribute"
-	GetActiveClusterInfoByWorkflowOpName             = "GetActiveClusterInfoByWorkflow"
-	GetActiveClusterSelectionPolicyForWorkflowOpName = "GetActiveClusterSelectionPolicyForWorkflow"
-	DomainIDToDomainFnErrorReason                    = "domain_id_to_name_fn_error"
+	LookupNewWorkflowOpName                                 = "LookupNewWorkflow"
+	LookupWorkflowOpName                                    = "LookupWorkflow"
+	GetActiveClusterInfoByClusterAttributeOpName            = "GetActiveClusterInfoByClusterAttribute"
+	GetActiveClusterInfoByWorkflowOpName                    = "GetActiveClusterInfoByWorkflow"
+	GetActiveClusterSelectionPolicyForWorkflowOpName        = "GetActiveClusterSelectionPolicyForWorkflow"
+	GetActiveClusterSelectionPolicyForCurrentWorkflowOpName = "GetActiveClusterSelectionPolicyForCurrentWorkflow"
+	DomainIDToDomainFnErrorReason                           = "domain_id_to_name_fn_error"
 
 	workflowPolicyCacheTTL      = 10 * time.Second
 	workflowPolicyCacheMaxCount = 1000
@@ -260,3 +261,40 @@ func (m *managerImpl) GetActiveClusterSelectionPolicyForWorkflow(ctx context.Con
 	}
 	return plcy, nil
 }
+
+func (m *managerImpl) GetActiveClusterSelectionPolicyForCurrentWorkflow(ctx context.Context, domainID, wfID string) (res *types.ActiveClusterSelectionPolicy, running bool, e error) {
+	d, scope, err := m.getDomainAndScope(domainID, GetActiveClusterSelectionPolicyForCurrentWorkflowOpName)
+	if err != nil {
+		return nil, false, err
+	}
+	defer m.handleError(scope, &e, time.Now())
+	if !d.GetReplicationConfig().IsActiveActive() {
+		// Not an active-active domain. return nil
+		m.logger.Debug("GetActiveClusterSelectionPolicyForCurrentWorkflow: not an active-active domain. returning nil",
+			tag.WorkflowDomainID(domainID),
+			tag.WorkflowID(wfID),
+		)
+		return nil, false, nil
+	}
+
+	shardID := common.WorkflowIDToHistoryShard(wfID, m.numShards)
+	executionManager, err := m.executionManagerProvider.GetExecutionManager(shardID)
+	if err != nil {
+		return nil, false, err
+	}
+	execution, err := executionManager.GetCurrentExecution(ctx, &persistence.GetCurrentExecutionRequest{
+		DomainID:   domainID,
+		WorkflowID: wfID,
+	})
+	if err != nil {
+		return nil, false, err
+	}
+	if persistence.IsWorkflowRunning(execution.State) {
+		policy, err := m.getClusterSelectionPolicy(ctx, domainID, wfID, execution.RunID)
+		if err != nil {
+			return nil, false, err
+		}
+		return policy, true, nil
+	}
+	return nil, false, nil
+}