Remove passive task resend logic

common/metrics/metric_defs.go

@@ -533,6 +533,8 @@ const (
 	BackfillHistoryEventsTaskScope = "BackfillHistoryEventsTask"
 	// VerifyVersionedTransitionTaskScope is the scope used by verify versioned transition task processing
 	VerifyVersionedTransitionTaskScope = "VerifyVersionedTransitionTask"
+	// SyncVersionedTransitionTaskScope is the scope used by sync versioned transition task processing
+	SyncVersionedTransitionTaskScope = "SyncVersionedTransitionTask"
 	// SyncWatermarkScope is the scope used by closed workflow task replication processing
 	SyncWatermarkScope = "SyncWatermark"
 	// NoopTaskScope is the scope used by noop task

service/history/ndc_standby_task_util.go

@@ -29,13 +29,16 @@ import (
 	"errors"
 	"time"
 
+	commonpb "go.temporal.io/api/common/v1"
 	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	"go.temporal.io/server/api/adminservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
+	"go.temporal.io/server/client"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/namespace"
-	"go.temporal.io/server/common/persistence/versionhistory"
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/tasks"
 	"go.temporal.io/server/service/history/workflow"
@@ -97,74 +100,81 @@ func standbyTimerTaskPostActionTaskDiscarded(
 	return consts.ErrTaskDiscarded
 }
 
-type (
-	historyResendInfo struct {
-
-		// used by NDC
-		lastEventID      int64
-		lastEventVersion int64
+func isWorkflowExistOnSource(
+	ctx context.Context,
+	taskInfo tasks.Task,
+	logger log.Logger,
+	currentCluster string,
+	clientBean client.Bean,
+	registry namespace.Registry,
+) bool {
+	workflowKey := taskWorkflowKey(taskInfo)
+	remoteClusterName, err := getRemoteClusterName(
+		currentCluster,
+		registry,
+		workflowKey.GetNamespaceID(),
+	)
+	if err != nil {
+		return true
 	}
+	remoteAdminClient, err := clientBean.GetRemoteAdminClient(remoteClusterName)
+	if err != nil {
+		return true
+	}
+	_, err = remoteAdminClient.DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{
+		Namespace: workflowKey.GetNamespaceID(),
+		Execution: &commonpb.WorkflowExecution{
+			WorkflowId: workflowKey.GetWorkflowID(),
+			RunId:      workflowKey.GetRunID(),
+		},
+	})
+	if err != nil {
+		if common.IsNotFoundError(err) {
+			return false
+		}
+		logger.Error("Error describe mutable state from remote.",
+			tag.WorkflowNamespaceID(workflowKey.GetNamespaceID()),
+			tag.WorkflowID(workflowKey.GetWorkflowID()),
+			tag.WorkflowRunID(workflowKey.GetRunID()),
+			tag.ClusterName(remoteClusterName),
+			tag.Error(err))
+	}
+	return true
+}
 
+type (
 	executionTimerPostActionInfo struct {
-		*historyResendInfo
-
 		currentRunID string
 	}
 
 	activityTaskPostActionInfo struct {
-		*historyResendInfo
-
 		taskQueue                          string
 		activityTaskScheduleToStartTimeout time.Duration
 		versionDirective                   *taskqueuespb.TaskVersionDirective
 	}
 
 	workflowTaskPostActionInfo struct {
-		*historyResendInfo
-
 		workflowTaskScheduleToStartTimeout time.Duration
 		taskqueue                          *taskqueuepb.TaskQueue
 		versionDirective                   *taskqueuespb.TaskVersionDirective
 	}
 )
 
-func newHistoryResendInfo(
-	lastEventID int64,
-	lastEventVersion int64,
-) *historyResendInfo {
-	return &historyResendInfo{
-		lastEventID:      lastEventID,
-		lastEventVersion: lastEventVersion,
-	}
-}
-
 func newExecutionTimerPostActionInfo(
 	mutableState workflow.MutableState,
 ) (*executionTimerPostActionInfo, error) {
-	resendInfo, err := getHistoryResendInfo(mutableState)
-	if err != nil {
-		return nil, err
-	}
-
 	return &executionTimerPostActionInfo{
-		historyResendInfo: resendInfo,
-		currentRunID:      mutableState.GetExecutionState().RunId,
+		currentRunID: mutableState.GetExecutionState().RunId,
 	}, nil
 }
 
 func newActivityTaskPostActionInfo(
 	mutableState workflow.MutableState,
 	activityInfo *persistencespb.ActivityInfo,
 ) (*activityTaskPostActionInfo, error) {
-	resendInfo, err := getHistoryResendInfo(mutableState)
-	if err != nil {
-		return nil, err
-	}
-
 	directive := MakeDirectiveForActivityTask(mutableState, activityInfo)
 
 	return &activityTaskPostActionInfo{
-		historyResendInfo:                  resendInfo,
 		activityTaskScheduleToStartTimeout: activityInfo.ScheduleToStartTimeout.AsDuration(),
 		versionDirective:                   directive,
 	}, nil
@@ -176,15 +186,9 @@ func newActivityRetryTimePostActionInfo(
 	activityScheduleToStartTimeout time.Duration,
 	activityInfo *persistencespb.ActivityInfo,
 ) (*activityTaskPostActionInfo, error) {
-	resendInfo, err := getHistoryResendInfo(mutableState)
-	if err != nil {
-		return nil, err
-	}
-
 	directive := MakeDirectiveForActivityTask(mutableState, activityInfo)
 
 	return &activityTaskPostActionInfo{
-		historyResendInfo:                  resendInfo,
 		taskQueue:                          taskQueue,
 		activityTaskScheduleToStartTimeout: activityScheduleToStartTimeout,
 		versionDirective:                   directive,
@@ -196,59 +200,30 @@ func newWorkflowTaskPostActionInfo(
 	workflowTaskScheduleToStartTimeout time.Duration,
 	taskqueue *taskqueuepb.TaskQueue,
 ) (*workflowTaskPostActionInfo, error) {
-	resendInfo, err := getHistoryResendInfo(mutableState)
-	if err != nil {
-		return nil, err
-	}
-
 	directive := MakeDirectiveForWorkflowTask(mutableState)
 
 	return &workflowTaskPostActionInfo{
-		historyResendInfo:                  resendInfo,
 		workflowTaskScheduleToStartTimeout: workflowTaskScheduleToStartTimeout,
 		taskqueue:                          taskqueue,
 		versionDirective:                   directive,
 	}, nil
 }
 
-func getHistoryResendInfo(
-	mutableState workflow.MutableState,
-) (*historyResendInfo, error) {
-
-	currentBranch, err := versionhistory.GetCurrentVersionHistory(mutableState.GetExecutionInfo().GetVersionHistories())
-	if err != nil {
-		return nil, err
-	}
-	lastItem, err := versionhistory.GetLastVersionHistoryItem(currentBranch)
-	if err != nil {
-		return nil, err
-	}
-	return newHistoryResendInfo(lastItem.GetEventId(), lastItem.GetVersion()), nil
-}
-
 func getStandbyPostActionFn(
 	taskInfo tasks.Task,
 	standbyNow standbyCurrentTimeFn,
-	standbyTaskMissingEventsResendDelay time.Duration,
 	standbyTaskMissingEventsDiscardDelay time.Duration,
-	fetchHistoryStandbyPostActionFn standbyPostActionFn,
 	discardTaskStandbyPostActionFn standbyPostActionFn,
 ) standbyPostActionFn {
 
 	// this is for task retry, use machine time
 	now := standbyNow()
 	taskTime := taskInfo.GetVisibilityTime()
-	resendTime := taskTime.Add(standbyTaskMissingEventsResendDelay)
 	discardTime := taskTime.Add(standbyTaskMissingEventsDiscardDelay)
 
 	// now < task start time + StandbyTaskMissingEventsResendDelay
-	if now.Before(resendTime) {
-		return standbyTaskPostActionNoOp
-	}
-
-	// task start time + StandbyTaskMissingEventsResendDelay <= now < task start time + StandbyTaskMissingEventsResendDelay
 	if now.Before(discardTime) {
-		return fetchHistoryStandbyPostActionFn
+		return standbyTaskPostActionNoOp
 	}
 
 	// task start time + StandbyTaskMissingEventsResendDelay <= now

service/history/replication/executable_sync_versioned_transition_task.go

@@ -66,7 +66,7 @@ func NewExecutableSyncVersionedTransitionTask(
 		ExecutableTask: NewExecutableTask(
 			processToolBox,
 			taskID,
-			metrics.VerifyVersionedTransitionTaskScope,
+			metrics.SyncVersionedTransitionTaskScope,
 			taskCreationTime,
 			time.Now().UTC(),
 			sourceClusterName,
@@ -102,7 +102,7 @@ func (e *ExecutableSyncVersionedTransitionTask) Execute() error {
 		)
 		metrics.ReplicationTasksSkipped.With(e.MetricsHandler).Record(
 			1,
-			metrics.OperationTag(metrics.VerifyVersionedTransitionTaskScope),
+			metrics.OperationTag(metrics.SyncVersionedTransitionTaskScope),
 			metrics.NamespaceTag(namespaceName),
 		)
 		return nil

service/history/timer_queue_factory.go

@@ -25,22 +25,14 @@
 package history
 
 import (
-	"context"
-
-	historypb "go.temporal.io/api/history/v1"
-	historyspb "go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/client"
-	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
-	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/resource"
-	"go.temporal.io/server/common/xdc"
 	deletemanager "go.temporal.io/server/service/history/deletemanager"
 	"go.temporal.io/server/service/history/queues"
-	"go.temporal.io/server/service/history/replication/eventhandler"
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/tasks"
 	wcache "go.temporal.io/server/service/history/workflow/cache"
@@ -146,68 +138,11 @@ func (f *timerQueueFactory) CreateQueue(
 		f.Config,
 		f.MatchingRawClient,
 	)
-	eventImporter := eventhandler.NewEventImporter(
-		f.RemoteHistoryFetcher,
-		func(ctx context.Context, namespaceID namespace.ID, workflowID string) (shard.Engine, error) {
-			return shardContext.GetEngine(ctx)
-		},
-		f.Serializer,
-		logger,
-	)
-	resendHandler := eventhandler.NewResendHandler(
-		f.NamespaceRegistry,
-		f.ClientBean,
-		f.Serializer,
-		f.ClusterMetadata,
-		func(ctx context.Context, namespaceID namespace.ID, workflowID string) (shard.Engine, error) {
-			return shardContext.GetEngine(ctx)
-		},
-		f.RemoteHistoryFetcher,
-		eventImporter,
-		logger,
-		f.Config,
-	)
 
 	standbyExecutor := newTimerQueueStandbyTaskExecutor(
 		shardContext,
 		workflowCache,
 		workflowDeleteManager,
-		xdc.NewNDCHistoryResender(
-			shardContext.GetNamespaceRegistry(),
-			f.ClientBean,
-			func(
-				ctx context.Context,
-				sourceClusterName string,
-				namespaceId namespace.ID,
-				workflowId string,
-				runId string,
-				events [][]*historypb.HistoryEvent,
-				versionHistory []*historyspb.VersionHistoryItem,
-			) error {
-				engine, err := shardContext.GetEngine(ctx)
-				if err != nil {
-					return err
-				}
-				return engine.ReplicateHistoryEvents(
-					ctx,
-					definition.WorkflowKey{
-						NamespaceID: namespaceId.String(),
-						WorkflowID:  workflowId,
-						RunID:       runId,
-					},
-					nil,
-					versionHistory,
-					events,
-					nil,
-					"",
-				)
-			},
-			shardContext.GetPayloadSerializer(),
-			f.Config.StandbyTaskReReplicationContextTimeout,
-			logger,
-			f.Config,
-		),
-		resendHandler,
 		f.MatchingRawClient,
 		logger,
 		f.MetricsHandler,
@@ -217,6 +152,7 @@ func (f *timerQueueFactory) CreateQueue(
 		// we have the option of revert to old behavior
 		currentClusterName,
 		f.Config,
+		f.ClientBean,
 	)
 
 	executor := queues.NewActiveStandbyExecutor(