De-duplicate reapply event (#2629)

* Dedup reapply event

* rename function

Author: yux0

service/history/historyEngine.go

@@ -50,6 +50,7 @@ import (
 	"go.temporal.io/server/service/history/tasks"
 
 	enumsspb "go.temporal.io/server/api/enums/v1"
+	historyspb "go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
@@ -78,7 +79,6 @@ import (
 	"go.temporal.io/server/service/history/events"
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/workflow"
-
 	"go.temporal.io/server/service/worker/archiver"
 )
 
@@ -3194,13 +3194,23 @@ func (e *historyEngineImpl) ReapplyEvents(
 		ctx,
 		namespaceID,
 		currentExecution,
-		func(context workflow.Context, mutableState workflow.MutableState) (*updateWorkflowAction, error) {
+		func(context workflow.Context, mutableState workflow.MutableState) (action *updateWorkflowAction, retErr error) {
 			// Filter out reapply event from the same cluster
 			toReapplyEvents := make([]*historypb.HistoryEvent, 0, len(reapplyEvents))
 			lastWriteVersion, err := mutableState.GetLastWriteVersion()
 			if err != nil {
 				return nil, err
 			}
+			sourceMutableState := mutableState
+			if sourceMutableState.GetWorkflowKey().RunID != runID {
+				originCtx, err := e.loadWorkflowOnce(ctx, namespaceID, workflowID, runID)
+				if err != nil {
+					return nil, err
+				}
+				defer func() { originCtx.getReleaseFn()(retErr) }()
+				sourceMutableState = originCtx.getMutableState()
+			}
+
 			for _, event := range reapplyEvents {
 				if event.GetVersion() == lastWriteVersion {
 					// The reapply is from the same cluster. Ignoring.
@@ -3211,6 +3221,11 @@ func (e *historyEngineImpl) ReapplyEvents(
 					// already apply the signal
 					continue
 				}
+				versionHistories := sourceMutableState.GetExecutionInfo().GetVersionHistories()
+				if e.containsHistoryEvent(versionHistories, event.GetEventId(), event.GetVersion()) {
+					continue
+				}
+
 				toReapplyEvents = append(toReapplyEvents, event)
 			}
 			if len(toReapplyEvents) == 0 {
@@ -3605,6 +3620,20 @@ func (e *historyEngineImpl) metricsScope(ctx context.Context) metrics.Scope {
 	return interceptor.MetricsScope(ctx, e.logger)
 }
 
+func (e *historyEngineImpl) containsHistoryEvent(
+	versionHistories *historyspb.VersionHistories,
+	reappliedEventID int64,
+	reappliedEventVersion int64,
+) bool {
+	// Check if the source workflow contains the reapply event.
+	// If it does, it means the event is received in this cluster, no need to reapply.
+	_, err := versionhistory.FindFirstVersionHistoryIndexByVersionHistoryItem(
+		versionHistories,
+		versionhistory.NewVersionHistoryItem(reappliedEventID, reappliedEventVersion),
+	)
+	return err == nil
+}
+
 func getMetadataChangeCallbackID(componentName string, shardId int32) string {
 	return fmt.Sprintf("%s-%d", componentName, shardId)
 }