Modify state builder to use standardlized task generation logic (cade…

…nce-workflow#2721)

* Rewrite state builder
* Rewrite UT

service/history/conflictResolver.go

@@ -123,10 +123,17 @@ func (r *conflictResolverImpl) reset(
 				domainEntry,
 			)
 
-			sBuilder = newStateBuilder(r.shard, resetMutableStateBuilder, r.logger)
+			sBuilder = newStateBuilder(
+				r.shard,
+				r.logger,
+				resetMutableStateBuilder,
+				func(mutableState mutableState) mutableStateTaskGenerator {
+					return newMutableStateTaskGenerator(r.shard.GetDomainCache(), r.logger, mutableState)
+				},
+			)
 		}
 
-		_, _, _, err = sBuilder.applyEvents(domainID, requestID, execution, history, nil, false)
+		_, err = sBuilder.applyEvents(domainID, requestID, execution, history, nil, false)
 		if err != nil {
 			r.logError("Conflict resolution err applying events.", err)
 			return nil, err

service/history/historyReplicator.go

@@ -130,7 +130,14 @@ func newHistoryReplicator(
 			return newConflictResolver(shard, context, historyV2Mgr, logger)
 		},
 		getNewStateBuilder: func(msBuilder mutableState, logger log.Logger) stateBuilder {
-			return newStateBuilder(shard, msBuilder, logger)
+			return newStateBuilder(
+				shard,
+				logger,
+				msBuilder,
+				func(mutableState mutableState) mutableStateTaskGenerator {
+					return newMutableStateTaskGenerator(shard.GetDomainCache(), logger, mutableState)
+				},
+			)
 		},
 		getNewMutableState: func(domainEntry *cache.DomainCacheEntry, logger log.Logger) mutableState {
 			return newMutableStateBuilderWithReplicationState(
@@ -565,6 +572,7 @@ func (r *historyReplicator) ApplyReplicationTask(
 	if len(request.History.Events) == 0 {
 		return nil
 	}
+	lastEvent := request.History.Events[len(request.History.Events)-1]
 
 	execution := *request.WorkflowExecution
 
@@ -576,7 +584,7 @@ func (r *historyReplicator) ApplyReplicationTask(
 	}
 
 	// directly use stateBuilder to apply events for other events(including continueAsNew)
-	lastEvent, _, newMutableState, err := sBuilder.applyEvents(
+	newMutableState, err := sBuilder.applyEvents(
 		domainID, requestID, execution, request.History.Events, newRunHistory, request.GetNewRunNDC(),
 	)
 	if err != nil {

service/history/mutableStateTaskGenerator.go

@@ -393,12 +393,13 @@ func (r *mutableStateTaskGeneratorImpl) generateRequestCancelExternalTasks(
 
 	attr := event.RequestCancelExternalWorkflowExecutionInitiatedEventAttributes
 	scheduleID := event.GetEventId()
+	version := event.GetVersion()
 	targetDomainName := attr.GetDomain()
 	targetWorkflowID := attr.GetWorkflowExecution().GetWorkflowId()
 	targetRunID := attr.GetWorkflowExecution().GetRunId()
 	targetChildOnly := attr.GetChildWorkflowOnly()
 
-	requestCancelExternalInfo, ok := r.mutableState.GetRequestCancelInfo(scheduleID)
+	_, ok := r.mutableState.GetRequestCancelInfo(scheduleID)
 	if !ok {
 		return &shared.InternalServiceError{
 			Message: fmt.Sprintf("it could be a bug, cannot get pending request cancel external workflow: %v", scheduleID),
@@ -418,7 +419,7 @@ func (r *mutableStateTaskGeneratorImpl) generateRequestCancelExternalTasks(
 		TargetRunID:             targetRunID,
 		TargetChildWorkflowOnly: targetChildOnly,
 		InitiatedID:             scheduleID,
-		Version:                 requestCancelExternalInfo.Version,
+		Version:                 version,
 	})
 
 	return nil
@@ -431,12 +432,13 @@ func (r *mutableStateTaskGeneratorImpl) generateSignalExternalTasks(
 
 	attr := event.SignalExternalWorkflowExecutionInitiatedEventAttributes
 	scheduleID := event.GetEventId()
+	version := event.GetVersion()
 	targetDomainName := attr.GetDomain()
 	targetWorkflowID := attr.GetWorkflowExecution().GetWorkflowId()
 	targetRunID := attr.GetWorkflowExecution().GetRunId()
 	targetChildOnly := attr.GetChildWorkflowOnly()
 
-	signalExternalInfo, ok := r.mutableState.GetSignalInfo(scheduleID)
+	_, ok := r.mutableState.GetSignalInfo(scheduleID)
 	if !ok {
 		return &shared.InternalServiceError{
 			Message: fmt.Sprintf("it could be a bug, cannot get pending signal external workflow: %v", scheduleID),
@@ -456,7 +458,7 @@ func (r *mutableStateTaskGeneratorImpl) generateSignalExternalTasks(
 		TargetRunID:             targetRunID,
 		TargetChildWorkflowOnly: targetChildOnly,
 		InitiatedID:             scheduleID,
-		Version:                 signalExternalInfo.Version,
+		Version:                 version,
 	})
 
 	return nil
@@ -471,7 +473,7 @@ func (r *mutableStateTaskGeneratorImpl) generateWorkflowSearchAttrTasks(
 	r.mutableState.AddTransferTasks(&persistence.UpsertWorkflowSearchAttributesTask{
 		// TaskID is set by shard
 		VisibilityTimestamp: now,
-		Version:             currentVersion,
+		Version:             currentVersion, // task processing does not check this version
 	})
 
 	return nil

service/history/nDCHistoryReplicator.go

@@ -46,7 +46,7 @@ var (
 
 type (
 	stateBuilderProvider func(
-		msBuilder mutableState,
+		mutableState mutableState,
 		logger log.Logger) stateBuilder
 
 	mutableStateProvider func(
@@ -149,10 +149,18 @@ func newNDCHistoryReplicator(
 			return newNDCWorkflowResetter(shard, transactionMgr, domainID, workflowID, baseRunID, newContext, newRunID, logger)
 		},
 		newStateBuilder: func(
-			msBuilder mutableState,
+			state mutableState,
 			logger log.Logger,
 		) stateBuilder {
-			return newStateBuilder(shard, msBuilder, logger)
+
+			return newStateBuilder(
+				shard,
+				logger,
+				state,
+				func(mutableState mutableState) mutableStateTaskGenerator {
+					return newMutableStateTaskGenerator(shard.GetDomainCache(), logger, mutableState)
+				},
+			)
 		},
 		newMutableState: func(
 			domainEntry *cache.DomainCacheEntry,
@@ -274,7 +282,7 @@ func (r *nDCHistoryReplicatorImpl) applyStartEvents(
 	stateBuilder := r.newStateBuilder(mutableState, task.getLogger())
 
 	// use state builder for workflow mutable state mutation
-	_, _, _, err = stateBuilder.applyEvents(
+	_, err = stateBuilder.applyEvents(
 		task.getDomainID(),
 		requestID,
 		*task.getExecution(),
@@ -354,7 +362,7 @@ func (r *nDCHistoryReplicatorImpl) applyNonStartEventsToCurrentBranch(
 
 	requestID := uuid.New() // requestID used for start workflow execution request.  This is not on the history event.
 	stateBuilder := r.newStateBuilder(mutableState, task.getLogger())
-	_, _, newMutableState, err := stateBuilder.applyEvents(
+	newMutableState, err := stateBuilder.applyEvents(
 		task.getDomainID(),
 		requestID,
 		*task.getExecution(),
@@ -523,7 +531,7 @@ func (r *nDCHistoryReplicatorImpl) applyNonStartEventsResetWorkflow(
 
 	requestID := uuid.New() // requestID used for start workflow execution request.  This is not on the history event.
 	stateBuilder := r.newStateBuilder(mutableState, task.getLogger())
-	_, _, _, err := stateBuilder.applyEvents(
+	_, err := stateBuilder.applyEvents(
 		task.getDomainID(),
 		requestID,
 		*task.getExecution(),

service/history/nDCStateRebuilder.go

@@ -185,7 +185,14 @@ func (r *nDCStateRebuilderImpl) initializeBuilders(
 		r.logger,
 		domainEntry,
 	)
-	stateBuilder := newStateBuilder(r.shard, resetMutableStateBuilder, r.logger)
+	stateBuilder := newStateBuilder(
+		r.shard,
+		r.logger,
+		resetMutableStateBuilder,
+		func(mutableState mutableState) mutableStateTaskGenerator {
+			return newMutableStateTaskGenerator(r.shard.GetDomainCache(), r.logger, mutableState)
+		},
+	)
 	return resetMutableStateBuilder, stateBuilder
 }
 
@@ -196,7 +203,7 @@ func (r *nDCStateRebuilderImpl) applyEvents(
 	requestID string,
 ) error {
 
-	_, _, _, err := stateBuilder.applyEvents(
+	_, err := stateBuilder.applyEvents(
 		workflowIdentifier.DomainID,
 		requestID,
 		shared.WorkflowExecution{

service/history/nDCStateRebuilder_test.go

@@ -164,7 +164,7 @@ func (s *nDCStateRebuilderSuite) TestApplyEvents() {
 		events,
 		[]*shared.HistoryEvent(nil),
 		true,
-	).Return(nil, nil, nil, nil).Times(1)
+	).Return(nil, nil).Times(1)
 
 	err := s.nDCStateRebuilder.applyEvents(workflowIdentifier, mockStateBuilder, events, requestID)
 	s.NoError(err)