fix(active-active): Fix active workflow check for history APIs (#7347)

<!-- Describe what has changed in this PR -->
**What changed?**
- Fix active workflow check for history APIs
- Update GetActiveClusterInfoByWorkflow to allow empty runID

<!-- Tell your future self why have you made these changes -->
**Why?**
To fix active cluster check for active-active domains

<!-- 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

common/activecluster/manager.go

@@ -33,6 +33,7 @@ import (
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
+	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/types"
 )
 
@@ -91,6 +92,21 @@ func NewManager(
 }
 
 func (m *managerImpl) getClusterSelectionPolicy(ctx context.Context, domainID, wfID, rID string) (*types.ActiveClusterSelectionPolicy, error) {
+	shardID := common.WorkflowIDToHistoryShard(wfID, m.numShards)
+	executionManager, err := m.executionManagerProvider.GetExecutionManager(shardID)
+	if err != nil {
+		return nil, err
+	}
+	if rID == "" {
+		execution, err := executionManager.GetCurrentExecution(ctx, &persistence.GetCurrentExecutionRequest{
+			DomainID:   domainID,
+			WorkflowID: wfID,
+		})
+		if err != nil {
+			return nil, err
+		}
+		rID = execution.RunID
+	}
 	// Check if the policy is already in the cache. create a key from domainID, wfID, rID
 	key := fmt.Sprintf("%s:%s:%s", domainID, wfID, rID)
 	cacheData := m.workflowPolicyCache.Get(key)
@@ -104,12 +120,6 @@ func (m *managerImpl) getClusterSelectionPolicy(ctx context.Context, domainID, w
 		m.logger.Warn(fmt.Sprintf("Cache data for key %s is of type %T, not a *types.ActiveClusterSelectionPolicy", key, cacheData))
 	}
 
-	shardID := common.WorkflowIDToHistoryShard(wfID, m.numShards)
-	executionManager, err := m.executionManagerProvider.GetExecutionManager(shardID)
-	if err != nil {
-		return nil, err
-	}
-
 	plcy, err := executionManager.GetActiveClusterSelectionPolicy(ctx, domainID, wfID, rID)
 	if err != nil {
 		return nil, err

common/activecluster/manager_test.go

@@ -304,6 +304,7 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 
 	tests := []struct {
 		name                           string
+		runID                          string
 		activeClusterCfg               *types.ActiveClusters
 		domainIDToNameErr              error
 		mockExecutionManagerFn         func(em *persistence.MockExecutionManager)
@@ -313,7 +314,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 		expectedError                  string
 	}{
 		{
-			name: "domain ID to name function returns error",
+			name:  "domain ID to name function returns error",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -330,7 +332,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			expectedError:     "failed to find domain by id",
 		},
 		{
-			name: "execution manager provider returns error",
+			name:  "execution manager provider returns error",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -349,7 +352,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			expectedError: "failed to get execution manager",
 		},
 		{
-			name: "execution manager GetActiveClusterSelectionPolicy returns error",
+			name:  "execution manager GetActiveClusterSelectionPolicy returns error",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -369,7 +373,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			expectedError: "database error",
 		},
 		{
-			name: "policy not found (EntityNotExistsError) - uses empty policy with nil cluster attribute",
+			name:  "policy not found (EntityNotExistsError) - uses empty policy with nil cluster attribute",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -392,7 +397,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			},
 		},
 		{
-			name: "policy not found (EntityNotExistsError) - empty policy with cluster attribute not found",
+			name:  "policy not found (EntityNotExistsError) - empty policy with cluster attribute not found",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -415,7 +421,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			},
 		},
 		{
-			name: "policy found but cluster attribute not found in domain config",
+			name:  "policy found but cluster attribute not found in domain config",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -440,7 +447,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			expectedError: "could not find cluster attribute &{datacenter dc1} in the domain test-domain-id's active cluster config",
 		},
 		{
-			name: "successful lookup - policy found and cluster attribute exists",
+			name:  "successful lookup - policy found and cluster attribute exists",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -472,7 +480,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			},
 		},
 		{
-			name: "successful lookup - policy from cache",
+			name:  "successful lookup - policy from cache",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"datacenter": {
@@ -495,14 +504,17 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 					Name:  "dc2",
 				},
 			},
-			// No mock needed since policy comes from cache
+			mockExecutionManagerFn: func(em *persistence.MockExecutionManager) {
+				// No expectations needed since policy comes from cache, but we need the provider to be non-nil
+			},
 			expectedResult: &types.ActiveClusterInfo{
 				ActiveClusterName: "cluster1",
 				FailoverVersion:   250,
 			},
 		},
 		{
-			name: "successful lookup - nil cluster attribute in policy uses domain-level info",
+			name:  "successful lookup - nil cluster attribute in policy uses domain-level info",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -527,7 +539,8 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			},
 		},
 		{
-			name: "successful lookup - multiple scopes with different attributes",
+			name:  "successful lookup - multiple scopes with different attributes",
+			runID: "test-run-id",
 			activeClusterCfg: &types.ActiveClusters{
 				AttributeScopes: map[string]types.ClusterAttributeScope{
 					"region": {
@@ -566,6 +579,71 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 				FailoverVersion:   300,
 			},
 		},
+		{
+			name:  "successful lookup - empty runID triggers GetCurrentExecution",
+			runID: "",
+			activeClusterCfg: &types.ActiveClusters{
+				AttributeScopes: map[string]types.ClusterAttributeScope{
+					"region": {
+						ClusterAttributes: map[string]types.ActiveClusterInfo{
+							"us-west": {
+								ActiveClusterName: "cluster0",
+								FailoverVersion:   100,
+							},
+							"us-east": {
+								ActiveClusterName: "cluster1",
+								FailoverVersion:   200,
+							},
+						},
+					},
+				},
+			},
+			mockExecutionManagerFn: func(em *persistence.MockExecutionManager) {
+				// Expect GetCurrentExecution to be called first to get the runID
+				em.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{
+					DomainID:   "test-domain-id",
+					WorkflowID: "test-workflow-id",
+				}).Return(&persistence.GetCurrentExecutionResponse{
+					RunID: "current-run-id",
+				}, nil)
+				// Then expect GetActiveClusterSelectionPolicy with the returned runID
+				em.EXPECT().GetActiveClusterSelectionPolicy(gomock.Any(), "test-domain-id", "test-workflow-id", "current-run-id").
+					Return(&types.ActiveClusterSelectionPolicy{
+						ClusterAttribute: &types.ClusterAttribute{
+							Scope: "region",
+							Name:  "us-east",
+						},
+					}, nil)
+			},
+			expectedResult: &types.ActiveClusterInfo{
+				ActiveClusterName: "cluster1",
+				FailoverVersion:   200,
+			},
+		},
+		{
+			name:  "empty runID - GetCurrentExecution returns error",
+			runID: "",
+			activeClusterCfg: &types.ActiveClusters{
+				AttributeScopes: map[string]types.ClusterAttributeScope{
+					"region": {
+						ClusterAttributes: map[string]types.ActiveClusterInfo{
+							"us-west": {
+								ActiveClusterName: "cluster0",
+								FailoverVersion:   100,
+							},
+						},
+					},
+				},
+			},
+			mockExecutionManagerFn: func(em *persistence.MockExecutionManager) {
+				// GetCurrentExecution returns an error
+				em.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{
+					DomainID:   "test-domain-id",
+					WorkflowID: "test-workflow-id",
+				}).Return(nil, errors.New("workflow not found"))
+			},
+			expectedError: "workflow not found",
+		},
 	}
 
 	for _, tc := range tests {
@@ -605,11 +683,11 @@ func TestGetActiveClusterInfoByWorkflow(t *testing.T) {
 			assert.NoError(t, err)
 
 			if tc.cachedPolicy != nil {
-				key := fmt.Sprintf("%s:%s:%s", "test-domain-id", wfID, "test-run-id")
+				key := fmt.Sprintf("%s:%s:%s", "test-domain-id", wfID, tc.runID)
 				mgr.(*managerImpl).workflowPolicyCache.Put(key, tc.cachedPolicy)
 			}
 
-			result, err := mgr.GetActiveClusterInfoByWorkflow(context.Background(), "test-domain-id", wfID, "test-run-id")
+			result, err := mgr.GetActiveClusterInfoByWorkflow(context.Background(), "test-domain-id", wfID, tc.runID)
 			if tc.expectedError != "" {
 				assert.EqualError(t, err, tc.expectedError)
 				assert.Nil(t, result)

common/cache/domainCache.go

@@ -1033,27 +1033,6 @@ func (entry *DomainCacheEntry) IsSampledForLongerRetention(
 	return false
 }
 
-// TODO(active-active): This function should accept active cluster selection policy as a parameter
-func GetActiveDomainByID(cache DomainCache, currentCluster string, domainID string) (*DomainCacheEntry, error) {
-	if err := common.ValidateDomainUUID(domainID); err != nil {
-		return nil, err
-	}
-
-	domain, err := cache.GetDomainByID(domainID)
-	if err != nil {
-		return nil, err
-	}
-
-	if !domain.IsActiveIn(currentCluster) {
-		// return the domain record as well as the not-active-error because some callers check
-		// whether the domain is pending active or not
-		// it's not a good design, but we need to keep it for backward compatibility
-		return domain, domain.NewDomainNotActiveError(currentCluster, domain.GetReplicationConfig().ActiveClusterName)
-	}
-
-	return domain, nil
-}
-
 // IsDeprecatedOrDeleted This function checks the domain status to see if the domain has been deprecated or deleted.
 func (entry *DomainCacheEntry) IsDeprecatedOrDeleted() bool {
 	if entry.info.Status == persistence.DomainStatusDeprecated || entry.info.Status == persistence.DomainStatusDeleted {

common/cache/domainCache_test.go

@@ -1117,74 +1117,6 @@ func Test_IsSampledForLongerRetention(t *testing.T) {
 	require.False(t, d.IsSampledForLongerRetention(wid))
 }
 
-func Test_GetActiveDomainByID(t *testing.T) {
-	nonExistingUUID := uuid.New()
-	activeDomainUUID := uuid.New()
-	passiveDomainUUID := uuid.New()
-
-	activeDomain := NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: activeDomainUUID, Name: "active"},
-		nil,
-		&persistence.DomainReplicationConfig{ActiveClusterName: "A"},
-		0,
-	)
-	passiveDomain := NewGlobalDomainCacheEntryForTest(
-		&persistence.DomainInfo{ID: passiveDomainUUID, Name: "passive"},
-		nil,
-		&persistence.DomainReplicationConfig{ActiveClusterName: "B"},
-		0,
-	)
-
-	tests := []struct {
-		msg          string
-		domainID     string
-		expectDomain *DomainCacheEntry
-		expectedErr  error
-	}{
-		{
-			msg:         "invalid UUID",
-			domainID:    "invalid",
-			expectedErr: &types.BadRequestError{Message: "Invalid domain UUID."},
-		},
-		{
-			msg:         "non existing domain",
-			domainID:    nonExistingUUID,
-			expectedErr: assert.AnError,
-		},
-		{
-			msg:          "active domain",
-			domainID:     activeDomainUUID,
-			expectDomain: activeDomain,
-		},
-		{
-			msg:          "passive domain",
-			domainID:     passiveDomainUUID,
-			expectDomain: passiveDomain,
-			expectedErr: &types.DomainNotActiveError{
-				Message:        "Domain: passive is active in cluster: B, while current cluster A is a standby cluster.",
-				DomainName:     "passive",
-				CurrentCluster: "A",
-				ActiveCluster:  "B",
-			},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.msg, func(t *testing.T) {
-			ctrl := gomock.NewController(t)
-			cache := NewMockDomainCache(ctrl)
-			cache.EXPECT().GetDomainByID(nonExistingUUID).Return(nil, assert.AnError).AnyTimes()
-			cache.EXPECT().GetDomainByID(activeDomainUUID).Return(activeDomain, nil).AnyTimes()
-			cache.EXPECT().GetDomainByID(passiveDomainUUID).Return(passiveDomain, nil).AnyTimes()
-
-			domain, err := GetActiveDomainByID(cache, "A", tt.domainID)
-
-			assert.Equal(t, tt.expectDomain, domain)
-			assert.Equal(t, tt.expectedErr, err)
-		})
-	}
-}
-
 func Test_WithTimeSource(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

service/history/engine/engineimpl/history_engine.go

@@ -425,10 +425,6 @@ func getScheduleID(activityID string, mutableState execution.MutableState) (int6
 	return activityInfo.ScheduleID, nil
 }
 
-func (e *historyEngineImpl) getActiveDomainByID(id string) (*cache.DomainCacheEntry, error) {
-	return cache.GetActiveDomainByID(e.shard.GetDomainCache(), e.clusterMetadata.GetCurrentClusterName(), id)
-}
-
 func (e *historyEngineImpl) getActiveDomainByWorkflow(ctx context.Context, domainID, workflowID, runID string) (*cache.DomainCacheEntry, error) {
 	activeClusterInfo, err := e.activeClusterManager.GetActiveClusterInfoByWorkflow(ctx, domainID, workflowID, runID)
 	if err != nil {