refactor(active-active): Refactor mutable state to accept currentVersion as input

service/history/engine/engineimpl/start_workflow_execution.go

@@ -657,19 +657,20 @@ UpdateWorkflowLoop:
 			return nil, err
 		}
 
-		// new mutable state
-		newMutableState, err := e.createMutableState(ctx, domainEntry, workflowExecution.GetRunID(), startRequest)
-		if err != nil {
-			return nil, err
-		}
-
+		var err error
 		if signalWithStartRequest != nil {
 			startRequest, err = getStartRequest(domainID, signalWithStartRequest.SignalWithStartRequest, signalWithStartRequest.PartitionConfig)
 			if err != nil {
 				return nil, err
 			}
 		}
 
+		// new mutable state
+		newMutableState, err := e.createMutableState(ctx, domainEntry, workflowExecution.GetRunID(), startRequest)
+		if err != nil {
+			return nil, err
+		}
+
 		err = e.addStartEventsAndTasks(
 			newMutableState,
 			workflowExecution,
@@ -847,25 +848,27 @@ func (e *historyEngineImpl) createMutableState(
 	runID string,
 	startRequest *types.HistoryStartWorkflowExecutionRequest,
 ) (execution.MutableState, error) {
+	version := domainEntry.GetFailoverVersion()
+	// TODO(active-active): replace with cluster attributes
+	if domainEntry.GetReplicationConfig().IsActiveActive() {
+		res, err := e.shard.GetActiveClusterManager().LookupNewWorkflow(ctx, domainEntry.GetInfo().ID, startRequest.StartRequest.ActiveClusterSelectionPolicy)
+		if err != nil {
+			return nil, err
+		}
+		version = res.FailoverVersion
+	}
 
 	newMutableState := execution.NewMutableStateBuilderWithVersionHistories(
 		e.shard,
 		e.logger,
 		domainEntry,
+		version,
 	)
 
 	if err := newMutableState.SetHistoryTree(runID); err != nil {
 		return nil, err
 	}
 
-	if domainEntry.GetReplicationConfig().IsActiveActive() {
-		res, err := e.shard.GetActiveClusterManager().LookupNewWorkflow(ctx, domainEntry.GetInfo().ID, startRequest.StartRequest.ActiveClusterSelectionPolicy)
-		if err != nil {
-			return nil, err
-		}
-		newMutableState.UpdateCurrentVersion(res.FailoverVersion, true)
-	}
-
 	return newMutableState, nil
 }
 

service/history/execution/mutable_state_builder.go

@@ -181,13 +181,14 @@ func NewMutableStateBuilder(
 	logger log.Logger,
 	domainEntry *cache.DomainCacheEntry,
 ) MutableState {
-	return newMutableStateBuilder(shard, logger, domainEntry)
+	return newMutableStateBuilder(shard, logger, domainEntry, constants.EmptyVersion)
 }
 
 func newMutableStateBuilder(
 	shard shard.Context,
 	logger log.Logger,
 	domainEntry *cache.DomainCacheEntry,
+	currentVersion int64,
 ) *mutableStateBuilder {
 	s := &mutableStateBuilder{
 		updateActivityInfos:        make(map[int64]*persistence.ActivityInfo),
@@ -217,7 +218,7 @@ func newMutableStateBuilder(
 		pendingSignalRequestedIDs: make(map[string]struct{}),
 		deleteSignalRequestedIDs:  make(map[string]struct{}),
 
-		currentVersion:        domainEntry.GetFailoverVersion(),
+		currentVersion:        currentVersion,
 		hasBufferedEventsInDB: false,
 		stateInDB:             persistence.WorkflowStateVoid,
 		nextEventIDInDB:       0,
@@ -257,13 +258,17 @@ func newMutableStateBuilder(
 }
 
 // NewMutableStateBuilderWithVersionHistories creates mutable state builder with version history initialized
+// NOTE: currentVersion should be the failover version of the workflow, which is derived from domain metadata and
+// the active cluster selection policy of the workflow. For passive workflows, the currentVersion will be overridden by the event version,
+// so the input doesn't matter for them.
 func NewMutableStateBuilderWithVersionHistories(
 	shard shard.Context,
 	logger log.Logger,
 	domainEntry *cache.DomainCacheEntry,
+	currentVersion int64,
 ) MutableState {
 
-	s := newMutableStateBuilder(shard, logger, domainEntry)
+	s := newMutableStateBuilder(shard, logger, domainEntry, currentVersion)
 	s.versionHistories = persistence.NewVersionHistories(&persistence.VersionHistory{})
 	return s
 }
@@ -291,7 +296,7 @@ func NewMutableStateBuilderWithVersionHistoriesWithEventV2(
 	domainEntry *cache.DomainCacheEntry,
 ) MutableState {
 
-	msBuilder := NewMutableStateBuilderWithVersionHistories(shard, logger, domainEntry)
+	msBuilder := NewMutableStateBuilderWithVersionHistories(shard, logger, domainEntry, domainEntry.GetFailoverVersion())
 	err := msBuilder.UpdateCurrentVersion(version, false)
 	if err != nil {
 		logger.Error("update current version error", tag.Error(err))
@@ -1269,6 +1274,7 @@ func (e *mutableStateBuilder) AddContinueAsNewEvent(
 		e.shard,
 		e.logger,
 		e.domainEntry,
+		e.domainEntry.GetFailoverVersion(),
 	).(*mutableStateBuilder)
 
 	// New mutable state initializes `currentVersion` to domain's failover version.

service/history/execution/mutable_state_builder_methods_activity_test.go

@@ -68,7 +68,7 @@ func testMutableStateBuilder(t *testing.T) *mutableStateBuilder {
 	mockShard.Resource.MatchingClient.EXPECT().AddActivityTask(gomock.Any(), gomock.Any()).Return(&types.AddActivityTaskResponse{}, nil).AnyTimes()
 	mockShard.Resource.DomainCache.EXPECT().GetDomainID(constants.TestDomainName).Return(constants.TestDomainID, nil).AnyTimes()
 	mockShard.Resource.DomainCache.EXPECT().GetDomainByID(constants.TestDomainID).Return(&cache.DomainCacheEntry{}, nil).AnyTimes()
-	return newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry)
+	return newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry, constants.TestLocalDomainEntry.GetFailoverVersion())
 }
 
 func Test__AddActivityTaskScheduledEvent(t *testing.T) {

service/history/execution/mutable_state_builder_methods_child_workflow_test.go

@@ -739,6 +739,7 @@ func loadMutableState(t *testing.T, ctx *shard.TestContext, state *persistence.W
 	m := newMutableStateBuilder(ctx,
 		log.NewNoop(),
 		domain,
+		domain.GetFailoverVersion(),
 	)
 	err := m.Load(context.Background(), state)
 	assert.NoError(t, err)

service/history/execution/mutable_state_builder_test.go

@@ -117,7 +117,7 @@ func (s *mutableStateSuite) SetupTest() {
 
 	s.mockShard.Resource.DomainCache.EXPECT().GetDomainID(constants.TestDomainName).Return(constants.TestDomainID, nil).AnyTimes()
 
-	s.msBuilder = newMutableStateBuilder(s.mockShard, s.logger, constants.TestLocalDomainEntry)
+	s.msBuilder = newMutableStateBuilder(s.mockShard, s.logger, constants.TestLocalDomainEntry, constants.TestLocalDomainEntry.GetFailoverVersion())
 }
 
 func (s *mutableStateSuite) TearDownTest() {
@@ -1956,7 +1956,7 @@ func TestMutableStateBuilder_CopyToPersistence_roundtrip(t *testing.T) {
 		activeClusterManager := activecluster.NewMockManager(ctrl)
 		shardContext.EXPECT().GetActiveClusterManager().Return(activeClusterManager).AnyTimes()
 
-		msb := newMutableStateBuilder(shardContext, log.NewNoop(), constants.TestGlobalDomainEntry)
+		msb := newMutableStateBuilder(shardContext, log.NewNoop(), constants.TestGlobalDomainEntry, constants.TestGlobalDomainEntry.GetFailoverVersion())
 
 		msb.Load(context.Background(), execution)
 
@@ -3894,7 +3894,7 @@ func createMSBWithMocks(mockCache *events.MockCache, shardContext *shardCtx.Mock
 		domainEntry = constants.TestGlobalDomainEntry
 	}
 
-	msb := newMutableStateBuilder(shardContext, log.NewNoop(), domainEntry)
+	msb := newMutableStateBuilder(shardContext, log.NewNoop(), domainEntry, domainEntry.GetFailoverVersion())
 	return msb
 }
 

service/history/execution/mutable_state_decision_task_manager_test.go

@@ -61,7 +61,7 @@ func TestReplicateDecisionTaskCompletedEvent(t *testing.T) {
 	mockShard.Resource.DomainCache.EXPECT().GetDomainID(constants.TestDomainName).Return(constants.TestDomainID, nil).AnyTimes()
 
 	m := &mutableStateDecisionTaskManagerImpl{
-		msb: newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry),
+		msb: newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry, constants.TestLocalDomainEntry.GetFailoverVersion()),
 	}
 	eventType := types.EventTypeActivityTaskCompleted
 	e := &types.HistoryEvent{
@@ -77,7 +77,7 @@ func TestReplicateDecisionTaskCompletedEvent(t *testing.T) {
 	assert.NoError(t, err)
 
 	// test when config is nil
-	m.msb = newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry)
+	m.msb = newMutableStateBuilder(mockShard, logger, constants.TestLocalDomainEntry, constants.TestLocalDomainEntry.GetFailoverVersion())
 	m.msb.config = nil
 	err = m.ReplicateDecisionTaskCompletedEvent(e)
 	assert.NoError(t, err)

service/history/execution/mutable_state_util_test.go

@@ -343,7 +343,7 @@ func TestCreatePersistenceMutableState(t *testing.T) {
 	mockShardContext.EXPECT().GetDomainCache().Return(mockDomainCache)
 	mockShardContext.EXPECT().GetLogger().Return(logger).AnyTimes()
 
-	builder := newMutableStateBuilder(mockShardContext, logger, constants.TestLocalDomainEntry)
+	builder := newMutableStateBuilder(mockShardContext, logger, constants.TestLocalDomainEntry, constants.TestLocalDomainEntry.GetFailoverVersion())
 	builder.pendingActivityInfoIDs[0] = &persistence.ActivityInfo{}
 	builder.pendingTimerInfoIDs["some-key"] = &persistence.TimerInfo{}
 	builder.pendingSignalInfoIDs[0] = &persistence.SignalInfo{}

service/history/execution/state_builder.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/cluster"
+	"github.com/uber/cadence/common/constants"
 	"github.com/uber/cadence/common/errors"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
@@ -488,10 +489,12 @@ func (b *stateBuilderImpl) ApplyEvents(
 
 			// The length of newRunHistory can be zero in resend case
 			if len(newRunHistory) != 0 {
+				domainEntry := b.mutableState.GetDomainEntry()
 				newRunMutableStateBuilder = NewMutableStateBuilderWithVersionHistories(
 					b.shard,
 					b.logger,
-					b.mutableState.GetDomainEntry(),
+					domainEntry,
+					constants.EmptyVersion,
 				)
 				newRunStateBuilder := NewStateBuilder(b.shard, b.logger, newRunMutableStateBuilder)
 				newRunID := event.WorkflowExecutionContinuedAsNewEventAttributes.GetNewExecutionRunID()

service/history/execution/state_rebuilder.go

@@ -210,6 +210,7 @@ func (r *stateRebuilderImpl) initializeBuilders(
 		r.shard,
 		r.logger,
 		domainEntry,
+		constants.EmptyVersion,
 	)
 	stateBuilder := NewStateBuilder(
 		r.shard,

service/history/ndc/history_replicator.go

@@ -29,6 +29,7 @@ import (
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/cluster"
+	"github.com/uber/cadence/common/constants"
 	"github.com/uber/cadence/common/errors"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
@@ -286,6 +287,7 @@ func NewHistoryReplicator(
 				shard,
 				logger,
 				domainEntry,
+				constants.EmptyVersion,
 			)
 		},
 		newReplicationTaskFn:                                         newReplicationTask,
@@ -419,6 +421,7 @@ func applyStartEvents(
 		return err
 	}
 	requestID := uuid.New() // requestID used for start workflow execution request.  This is not on the history event.
+	// since it's replicated from the other cluster, we don't care the active cluster policy, because the failover version will be updated after ApplyEvents
 	mutableState := newMutableState(domainEntry, task.getLogger())
 	stateBuilder := newStateBuilder(mutableState, task.getLogger())