State Based replication fix (#7553)

## What changed?
<!-- Describe what has changed in this PR -->
1. Update the mutable state update execution state to ignore the change
when no change on the status
2. Update the caller of Partial refresh to provide inclusive state
transition.
3. Add events to event cache when processing sync versioned transition
task
## Why?
<!-- Tell your future self why have you made these changes -->
1 and 2 is to reduce task refresh count. 3 is for reduce
ReadHistoryBranch call.
## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
unit test
## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
no risk
## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->
n/a
## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->
yes

Author: xwduan

service/history/ndc/workflow_state_replicator.go

@@ -60,6 +60,7 @@ import (
 	"go.temporal.io/server/common/primitives/timestamp"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/service/history/consts"
+	"go.temporal.io/server/service/history/events"
 	"go.temporal.io/server/service/history/historybuilder"
 	historyi "go.temporal.io/server/service/history/interfaces"
 	"go.temporal.io/server/service/history/workflow"
@@ -404,8 +405,9 @@ func (r *WorkflowStateReplicatorImpl) applyMutation(
 			return err
 		}
 	}
-
-	err = r.taskRefresher.PartialRefresh(ctx, localMutableState, localVersionedTransition)
+	nextVersionedTransition := transitionhistory.CopyVersionedTransition(localVersionedTransition)
+	nextVersionedTransition.TransitionCount++
+	err = r.taskRefresher.PartialRefresh(ctx, localMutableState, nextVersionedTransition)
 	if err != nil {
 		return err
 	}
@@ -550,7 +552,9 @@ func (r *WorkflowStateReplicatorImpl) applySnapshotWhenWorkflowExist(
 			return err
 		}
 	} else {
-		err = r.taskRefresher.PartialRefresh(ctx, localMutableState, localVersionedTransition)
+		nextVersionedTransition := transitionhistory.CopyVersionedTransition(localVersionedTransition)
+		nextVersionedTransition.TransitionCount++
+		err = r.taskRefresher.PartialRefresh(ctx, localMutableState, nextVersionedTransition)
 		if err != nil {
 			return err
 		}
@@ -839,6 +843,7 @@ func (r *WorkflowStateReplicatorImpl) bringLocalEventsUpToSourceCurrentBranch(
 			}
 			for _, event := range events {
 				localMutableState.AddReapplyCandidateEvent(event)
+				r.addEventToCache(localMutableState.GetWorkflowKey(), event)
 			}
 			_, err = r.executionMgr.AppendRawHistoryNodes(ctx, &persistence.AppendRawHistoryNodesRequest{
 				ShardID:           r.shardContext.GetShardID(),
@@ -888,6 +893,7 @@ func (r *WorkflowStateReplicatorImpl) bringLocalEventsUpToSourceCurrentBranch(
 		}
 		for _, event := range events {
 			localMutableState.AddReapplyCandidateEvent(event)
+			r.addEventToCache(localMutableState.GetWorkflowKey(), event)
 		}
 		_, err = r.executionMgr.AppendRawHistoryNodes(ctx, &persistence.AppendRawHistoryNodesRequest{
 			ShardID:           r.shardContext.GetShardID(),
@@ -1202,6 +1208,7 @@ BackfillLoop:
 			}
 			for _, event := range events {
 				mutableState.AddReapplyCandidateEvent(event)
+				r.addEventToCache(mutableState.GetWorkflowKey(), event)
 			}
 		}
 
@@ -1348,6 +1355,37 @@ func (r *WorkflowStateReplicatorImpl) getHistoryFromRemotePaginationFn(
 	}
 }
 
+func (r *WorkflowStateReplicatorImpl) addEventToCache(
+	workflowKey definition.WorkflowKey,
+	event *historypb.HistoryEvent,
+) {
+	switch event.EventType {
+	case enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
+		enumspb.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED,
+		enumspb.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED,
+		enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED:
+		r.shardContext.GetEventsCache().PutEvent(
+			events.EventKey{
+				NamespaceID: namespace.ID(workflowKey.NamespaceID),
+				WorkflowID:  workflowKey.WorkflowID,
+				RunID:       workflowKey.RunID,
+				EventID:     event.GetEventId(),
+				Version:     event.GetVersion(),
+			},
+			event,
+		)
+	default:
+	}
+
+}
+
 func sortAncestors(ans []*persistencespb.HistoryBranchRange) []*persistencespb.HistoryBranchRange {
 	if len(ans) > 0 {
 		// sort ans based onf EndNodeID so that we can set BeginNodeID

service/history/ndc/workflow_state_replicator_test.go

@@ -695,7 +695,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_SameBranch_S
 	mockTaskRefresher.EXPECT().
 		PartialRefresh(gomock.Any(), gomock.Any(), EqVersionedTransition(&persistencespb.VersionedTransition{
 			NamespaceFailoverVersion: 2,
-			TransitionCount:          18,
+			TransitionCount:          19,
 		}),
 		).Return(nil).Times(1)
 
@@ -870,7 +870,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_SameBranch_S
 	mockTaskRefresher.EXPECT().
 		PartialRefresh(gomock.Any(), gomock.Any(), EqVersionedTransition(&persistencespb.VersionedTransition{
 			NamespaceFailoverVersion: 2,
-			TransitionCount:          18,
+			TransitionCount:          19,
 		}),
 		).Return(nil).Times(1)
 
@@ -1053,6 +1053,7 @@ func (s *workflowReplicatorSuite) Test_bringLocalEventsUpToSourceCurrentBranch_W
 	mockMutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{
 		RunId: s.runID,
 	}).AnyTimes()
+	mockMutableState.EXPECT().GetWorkflowKey().Return(definition.NewWorkflowKey(namespaceID, s.workflowID, s.runID)).AnyTimes()
 	mockMutableState.EXPECT().SetHistoryBuilder(gomock.Any()).Times(1)
 
 	allEvents := append(gapEvents, requestedEvents...)

service/history/workflow/mutable_state_impl.go

@@ -5754,6 +5754,9 @@ func (ms *MutableStateImpl) UpdateWorkflowStateStatus(
 	state enumsspb.WorkflowExecutionState,
 	status enumspb.WorkflowExecutionStatus,
 ) error {
+	if state == ms.executionState.State && status == ms.executionState.Status {
+		return nil
+	}
 	if state != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
 		ms.executionStateUpdated = true
 		ms.visibilityUpdated = true // workflow status & state change triggers visibility change as well

service/history/workflow/task_refresher.go

@@ -98,17 +98,6 @@ func (r *TaskRefresherImpl) PartialRefresh(
 	mutableState historyi.MutableState,
 	minVersionedTransition *persistencespb.VersionedTransition,
 ) error {
-	if transitionhistory.Compare(minVersionedTransition, EmptyVersionedTransition) != 0 {
-		// Perform a sanity check to make sure that the minVersionedTransition, if provided,
-		// is on the current branch of transition history.
-		if err := TransitionHistoryStalenessCheck(
-			mutableState.GetExecutionInfo().TransitionHistory,
-			minVersionedTransition,
-		); err != nil {
-			return err
-		}
-	}
-
 	taskGenerator := r.taskGeneratorProvider.NewTaskGenerator(
 		r.shard,
 		mutableState,
@@ -410,13 +399,8 @@ func (r *TaskRefresherImpl) refreshTasksForActivity(
 			continue
 		}
 
-		scheduleEvent, err := mutableState.GetActivityScheduledEvent(ctx, activityInfo.ScheduledEventId)
-		if err != nil {
-			return err
-		}
-
 		if err := taskGenerator.GenerateActivityTasks(
-			scheduleEvent.GetEventId(),
+			activityInfo.ScheduledEventId,
 		); err != nil {
 			return err
 		}
@@ -605,7 +589,6 @@ func (r *TaskRefresherImpl) refreshTasksForWorkflowSearchAttr(
 	) < 0 {
 		return nil
 	}
-
 	return taskGenerator.GenerateUpsertVisibilityTask()
 }
 

service/history/workflow/task_refresher_test.go

@@ -613,31 +613,6 @@ func (s *taskRefresherSuite) TestRefreshActivityTasks() {
 				10,
 			)
 			s.NoError(err)
-
-			for _, eventID := range tc.getActivityScheduledEventIDs {
-				// only the first activity will actually refresh the transfer activity task
-				scheduledEvent := &historypb.HistoryEvent{
-					EventId:   eventID,
-					Version:   common.EmptyVersion,
-					EventType: enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
-					Attributes: &historypb.HistoryEvent_ActivityTaskScheduledEventAttributes{
-						ActivityTaskScheduledEventAttributes: &historypb.ActivityTaskScheduledEventAttributes{},
-					},
-				}
-				s.mockShard.MockEventsCache.EXPECT().GetEvent(
-					gomock.Any(),
-					s.mockShard.GetShardID(),
-					events.EventKey{
-						NamespaceID: tests.NamespaceID,
-						WorkflowID:  tests.WorkflowID,
-						RunID:       tests.RunID,
-						EventID:     eventID,
-						Version:     common.EmptyVersion,
-					},
-					int64(4),
-					branchToken,
-				).Return(scheduledEvent, nil).Times(1)
-			}
 			for _, eventID := range tc.generateActivityTaskIDs {
 				s.mockTaskGenerator.EXPECT().GenerateActivityTasks(int64(eventID)).Return(nil).Times(1)
 			}