Extract workflow execution delete code to workflow.DeleteManager (#2388)

common/metrics/defs.go

@@ -1091,6 +1091,8 @@ const (
 	HistoryQueryWorkflowScope
 	// HistoryProcessDeleteHistoryEventScope tracks ProcessDeleteHistoryEvent processing calls
 	HistoryProcessDeleteHistoryEventScope
+	// HistoryDeleteWorkflowExecutionScope tracks DeleteWorkflowExecutions API calls
+	HistoryDeleteWorkflowExecutionScope
 	// WorkflowCompletionStatsScope tracks workflow completion updates
 	WorkflowCompletionStatsScope
 	// ArchiverClientScope is scope used by all metrics emitted by archiver.Client
@@ -1586,6 +1588,7 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 		HistoryResetWorkflowExecutionScope:              {operation: "ResetWorkflowExecution"},
 		HistoryQueryWorkflowScope:                       {operation: "QueryWorkflow"},
 		HistoryProcessDeleteHistoryEventScope:           {operation: "ProcessDeleteHistoryEvent"},
+		HistoryDeleteWorkflowExecutionScope:             {operation: "DeleteWorkflowExecution"},
 		HistoryScheduleWorkflowTaskScope:                {operation: "ScheduleWorkflowTask"},
 		HistoryRecordChildExecutionCompletedScope:       {operation: "RecordChildExecutionCompleted"},
 		HistoryRequestCancelWorkflowExecutionScope:      {operation: "RequestCancelWorkflowExecution"},

service/history/consts/const.go

@@ -54,6 +54,8 @@ var (
 	ErrActivityTaskNotCancelRequested = serviceerror.NewInvalidArgument("unable to mark activity as canceled without activity being request canceled first")
 	// ErrWorkflowCompleted is the error to indicate workflow execution already completed
 	ErrWorkflowCompleted = serviceerror.NewNotFound("workflow execution already completed")
+	// ErrWorkflowNotCompleted is the error to indicate workflow execution is not completed.
+	ErrWorkflowNotCompleted = serviceerror.NewNotFound("workflow execution is not completed")
 	// ErrWorkflowExecutionNotFound is the error to indicate workflow execution does not exist
 	ErrWorkflowExecutionNotFound = serviceerror.NewNotFound("workflow execution not found")
 	// ErrWorkflowParent is the error to parent execution is given and mismatch

service/history/historyEngine.go

@@ -120,7 +120,7 @@ type (
 		rawMatchingClient             matchingservice.MatchingServiceClient
 		replicationDLQHandler         replicationDLQHandler
 		searchAttributesValidator     *searchattribute.Validator
-		searchAttributesMapper        searchattribute.Mapper
+		workflowDeleteManager         workflow.DeleteManager
 	}
 )
 
@@ -145,33 +145,44 @@ func NewEngineWithShardContext(
 	logger := shard.GetLogger()
 	executionManager := shard.GetExecutionManager()
 	historyCache := newCacheFn(shard)
+
+	archivalClient := archiver.NewClient(
+		shard.GetMetricsClient(),
+		logger,
+		publicClient,
+		shard.GetConfig().NumArchiveSystemWorkflows,
+		shard.GetConfig().ArchiveRequestRPS,
+		archiverProvider,
+	)
+
+	workflowDeleteManager := workflow.NewDeleteManager(
+		shard,
+		historyCache,
+		config,
+		archivalClient,
+	)
+
 	historyEngImpl := &historyEngineImpl{
-		status:             common.DaemonStatusInitialized,
-		currentClusterName: currentClusterName,
-		shard:              shard,
-		clusterMetadata:    shard.GetClusterMetadata(),
-		timeSource:         shard.GetTimeSource(),
-		executionManager:   executionManager,
-		tokenSerializer:    common.NewProtoTaskTokenSerializer(),
-		historyCache:       historyCache,
-		logger:             log.With(logger, tag.ComponentHistoryEngine),
-		throttledLogger:    log.With(shard.GetThrottledLogger(), tag.ComponentHistoryEngine),
-		metricsClient:      shard.GetMetricsClient(),
-		eventNotifier:      eventNotifier,
-		config:             config,
-		archivalClient: archiver.NewClient(
-			shard.GetMetricsClient(),
-			logger,
-			publicClient,
-			shard.GetConfig().NumArchiveSystemWorkflows,
-			shard.GetConfig().ArchiveRequestRPS,
-			archiverProvider,
-		),
+		status:                    common.DaemonStatusInitialized,
+		currentClusterName:        currentClusterName,
+		shard:                     shard,
+		clusterMetadata:           shard.GetClusterMetadata(),
+		timeSource:                shard.GetTimeSource(),
+		executionManager:          executionManager,
+		tokenSerializer:           common.NewProtoTaskTokenSerializer(),
+		historyCache:              historyCache,
+		logger:                    log.With(logger, tag.ComponentHistoryEngine),
+		throttledLogger:           log.With(shard.GetThrottledLogger(), tag.ComponentHistoryEngine),
+		metricsClient:             shard.GetMetricsClient(),
+		eventNotifier:             eventNotifier,
+		config:                    config,
+		archivalClient:            archivalClient,
 		publicClient:              publicClient,
 		matchingClient:            matchingClient,
 		rawMatchingClient:         rawMatchingClient,
 		replicationTaskProcessors: make(map[string]ReplicationTaskProcessor),
 		replicationTaskFetchers:   replicationTaskFetchers,
+		workflowDeleteManager:     workflowDeleteManager,
 	}
 
 	historyEngImpl.txProcessor = newTransferQueueProcessor(shard, historyEngImpl,
@@ -217,8 +228,6 @@ func NewEngineWithShardContext(
 		config.SearchAttributesTotalSizeLimit,
 	)
 
-	historyEngImpl.searchAttributesMapper = shard.GetSearchAttributesMapper()
-
 	historyEngImpl.workflowTaskHandler = newWorkflowTaskHandlerCallback(historyEngImpl)
 	historyEngImpl.replicationDLQHandler = newLazyReplicationDLQHandler(shard)
 
@@ -514,7 +523,7 @@ func (e *historyEngineImpl) StartWorkflowExecution(
 		return nil, err
 	}
 
-	err = searchattribute.SubstituteAliases(e.searchAttributesMapper, request.GetSearchAttributes(), namespace.String())
+	err = searchattribute.SubstituteAliases(e.shard.GetSearchAttributesMapper(), request.GetSearchAttributes(), namespace.String())
 	if err != nil {
 		return nil, err
 	}
@@ -2012,7 +2021,7 @@ func (e *historyEngineImpl) SignalWithStartWorkflowExecution(
 		return nil, err
 	}
 
-	err = searchattribute.SubstituteAliases(e.searchAttributesMapper, request.GetSearchAttributes(), namespace.String())
+	err = searchattribute.SubstituteAliases(e.shard.GetSearchAttributesMapper(), request.GetSearchAttributes(), namespace.String())
 	if err != nil {
 		return nil, err
 	}

service/history/historyEngine2_test.go

@@ -144,21 +144,20 @@ func (s *engine2Suite) SetupTest() {
 
 	historyCache := workflow.NewCache(s.mockShard)
 	h := &historyEngineImpl{
-		currentClusterName:     s.mockShard.GetClusterMetadata().GetCurrentClusterName(),
-		shard:                  s.mockShard,
-		clusterMetadata:        s.mockClusterMetadata,
-		executionManager:       s.mockExecutionMgr,
-		historyCache:           historyCache,
-		logger:                 s.logger,
-		throttledLogger:        s.logger,
-		metricsClient:          metrics.NewNoopMetricsClient(),
-		tokenSerializer:        common.NewProtoTaskTokenSerializer(),
-		config:                 s.config,
-		timeSource:             s.mockShard.GetTimeSource(),
-		eventNotifier:          events.NewNotifier(clock.NewRealTimeSource(), metrics.NewNoopMetricsClient(), func(namespace.ID, string) int32 { return 1 }),
-		txProcessor:            s.mockTxProcessor,
-		timerProcessor:         s.mockTimerProcessor,
-		searchAttributesMapper: s.mockShard.Resource.SearchAttributesMapper,
+		currentClusterName: s.mockShard.GetClusterMetadata().GetCurrentClusterName(),
+		shard:              s.mockShard,
+		clusterMetadata:    s.mockClusterMetadata,
+		executionManager:   s.mockExecutionMgr,
+		historyCache:       historyCache,
+		logger:             s.logger,
+		throttledLogger:    s.logger,
+		metricsClient:      metrics.NewNoopMetricsClient(),
+		tokenSerializer:    common.NewProtoTaskTokenSerializer(),
+		config:             s.config,
+		timeSource:         s.mockShard.GetTimeSource(),
+		eventNotifier:      events.NewNotifier(clock.NewRealTimeSource(), metrics.NewNoopMetricsClient(), func(namespace.ID, string) int32 { return 1 }),
+		txProcessor:        s.mockTxProcessor,
+		timerProcessor:     s.mockTimerProcessor,
 		searchAttributesValidator: searchattribute.NewValidator(
 			searchattribute.NewTestProvider(),
 			s.mockShard.Resource.SearchAttributesMapper,

service/history/shard/context_impl.go

@@ -784,7 +784,7 @@ func (s *ContextImpl) AppendHistoryEvents(
 func (s *ContextImpl) DeleteWorkflowExecution(
 	key definition.WorkflowKey,
 	branchToken []byte,
-	version int64,
+	newTaskVersion int64,
 ) error {
 	if err := s.errorByState(); err != nil {
 		return err
@@ -855,7 +855,7 @@ func (s *ContextImpl) DeleteWorkflowExecution(
 			// TaskID is set by addTasksLocked
 			WorkflowKey:         key,
 			VisibilityTimestamp: s.timeSource.Now(),
-			Version:             version,
+			Version:             newTaskVersion,
 		}},
 	}
 	err = s.addTasksLocked(addTasksRequest, namespaceEntry)

service/history/shard/context_testutil.go

@@ -94,6 +94,7 @@ func NewTestContext(
 		persistenceShardManager: resource.GetShardManager(),
 		clientBean:              resource.GetClientBean(),
 		saProvider:              resource.GetSearchAttributesProvider(),
+		saMapper:                resource.GetSearchAttributesMapper(),
 		historyClient:           resource.GetHistoryClient(),
 		archivalMetadata:        resource.GetArchivalMetadata(),
 	}

service/history/shard/engine.go

@@ -32,6 +32,7 @@ import (
 
 	commonpb "go.temporal.io/api/common/v1"
 	historypb "go.temporal.io/api/history/v1"
+
 	"go.temporal.io/server/api/historyservice/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
 	"go.temporal.io/server/common"

service/history/timerQueueActiveProcessor.go

@@ -97,7 +97,8 @@ func newTimerQueueActiveProcessor(
 	}
 	processor.taskExecutor = newTimerQueueActiveTaskExecutor(
 		shard,
-		historyService,
+		historyService.workflowDeleteManager,
+		historyService.historyCache,
 		processor,
 		logger,
 		historyService.metricsClient,
@@ -189,7 +190,8 @@ func newTimerQueueFailoverProcessor(
 	}
 	processor.taskExecutor = newTimerQueueActiveTaskExecutor(
 		shard,
-		historyService,
+		historyService.workflowDeleteManager,
+		historyService.historyCache,
 		processor,
 		logger,
 		historyService.metricsClient,

service/history/timerQueueActiveTaskExecutor.go

@@ -62,7 +62,8 @@ type (
 
 func newTimerQueueActiveTaskExecutor(
 	shard shard.Context,
-	historyService *historyEngineImpl,
+	workflowDeleteManager workflow.DeleteManager,
+	cache workflow.Cache,
 	queueProcessor *timerQueueActiveProcessorImpl,
 	logger log.Logger,
 	metricsClient metrics.Client,
@@ -72,7 +73,8 @@ func newTimerQueueActiveTaskExecutor(
 	return &timerQueueActiveTaskExecutor{
 		timerQueueTaskExecutorBase: newTimerQueueTaskExecutorBase(
 			shard,
-			historyService,
+			workflowDeleteManager,
+			cache,
 			logger,
 			metricsClient,
 			config,

service/history/timerQueueActiveTaskExecutor_test.go

@@ -74,6 +74,7 @@ type (
 		mockClusterMetadata *cluster.MockMetadata
 
 		mockHistoryEngine *historyEngineImpl
+		mockDeleteManager *workflow.MockDeleteManager
 		mockExecutionMgr  *persistence.MockExecutionManager
 
 		logger                       log.Logger
@@ -147,25 +148,28 @@ func (s *timerQueueActiveTaskExecutorSuite) SetupTest() {
 	s.logger = s.mockShard.GetLogger()
 
 	historyCache := workflow.NewCache(s.mockShard)
+	s.mockDeleteManager = workflow.NewMockDeleteManager(s.controller)
 	h := &historyEngineImpl{
-		currentClusterName: s.mockShard.Resource.GetClusterMetadata().GetCurrentClusterName(),
-		shard:              s.mockShard,
-		clusterMetadata:    s.mockClusterMetadata,
-		executionManager:   s.mockExecutionMgr,
-		historyCache:       historyCache,
-		logger:             s.logger,
-		tokenSerializer:    common.NewProtoTaskTokenSerializer(),
-		metricsClient:      s.mockShard.GetMetricsClient(),
-		eventNotifier:      events.NewNotifier(clock.NewRealTimeSource(), metrics.NewNoopMetricsClient(), func(namespace.ID, string) int32 { return 1 }),
-		txProcessor:        s.mockTxProcessor,
-		timerProcessor:     s.mockTimerProcessor,
+		currentClusterName:    s.mockShard.Resource.GetClusterMetadata().GetCurrentClusterName(),
+		shard:                 s.mockShard,
+		clusterMetadata:       s.mockClusterMetadata,
+		executionManager:      s.mockExecutionMgr,
+		historyCache:          historyCache,
+		logger:                s.logger,
+		tokenSerializer:       common.NewProtoTaskTokenSerializer(),
+		metricsClient:         s.mockShard.GetMetricsClient(),
+		eventNotifier:         events.NewNotifier(clock.NewRealTimeSource(), metrics.NewNoopMetricsClient(), func(namespace.ID, string) int32 { return 1 }),
+		txProcessor:           s.mockTxProcessor,
+		timerProcessor:        s.mockTimerProcessor,
+		workflowDeleteManager: s.mockDeleteManager,
 	}
 	s.mockShard.SetEngineForTesting(h)
 	s.mockHistoryEngine = h
 
 	s.timerQueueActiveTaskExecutor = newTimerQueueActiveTaskExecutor(
 		s.mockShard,
-		h,
+		s.mockDeleteManager,
+		historyCache,
 		newTimerQueueActiveProcessor(
 			s.mockShard,
 			h,

service/history/timerQueueStandbyProcessor.go

@@ -97,7 +97,8 @@ func newTimerQueueStandbyProcessor(
 		timerGate:       timerGate,
 		taskExecutor: newTimerQueueStandbyTaskExecutor(
 			shard,
-			historyService,
+			historyService.workflowDeleteManager,
+			historyService.historyCache,
 			nDCHistoryResender,
 			logger,
 			historyService.metricsClient,

service/history/timerQueueStandbyTaskExecutor.go

@@ -60,7 +60,8 @@ type (
 
 func newTimerQueueStandbyTaskExecutor(
 	shard shard.Context,
-	historyService *historyEngineImpl,
+	deleteManager workflow.DeleteManager,
+	cache workflow.Cache,
 	nDCHistoryResender xdc.NDCHistoryResender,
 	logger log.Logger,
 	metricsClient metrics.Client,
@@ -71,7 +72,8 @@ func newTimerQueueStandbyTaskExecutor(
 	return &timerQueueStandbyTaskExecutor{
 		timerQueueTaskExecutorBase: newTimerQueueTaskExecutorBase(
 			shard,
-			historyService,
+			deleteManager,
+			cache,
 			logger,
 			metricsClient,
 			config,

service/history/timerQueueStandbyTaskExecutor_test.go

@@ -78,6 +78,7 @@ type (
 		mockClusterMetadata      *cluster.MockMetadata
 		mockAdminClient          *adminservicemock.MockAdminServiceClient
 		mockNDCHistoryResender   *xdc.MockNDCHistoryResender
+		mockDeleteManager        *workflow.MockDeleteManager
 
 		logger               log.Logger
 		namespaceID          namespace.ID
@@ -162,26 +163,29 @@ func (s *timerQueueStandbyTaskExecutorSuite) SetupTest() {
 	s.logger = s.mockShard.GetLogger()
 
 	historyCache := workflow.NewCache(s.mockShard)
+	s.mockDeleteManager = workflow.NewMockDeleteManager(s.controller)
 	h := &historyEngineImpl{
-		currentClusterName: s.mockShard.Resource.GetClusterMetadata().GetCurrentClusterName(),
-		shard:              s.mockShard,
-		clusterMetadata:    s.mockClusterMetadata,
-		executionManager:   s.mockExecutionMgr,
-		historyCache:       historyCache,
-		logger:             s.logger,
-		tokenSerializer:    common.NewProtoTaskTokenSerializer(),
-		metricsClient:      s.mockShard.GetMetricsClient(),
-		eventNotifier:      events.NewNotifier(s.timeSource, metrics.NewNoopMetricsClient(), func(namespace.ID, string) int32 { return 1 }),
-		txProcessor:        s.mockTxProcessor,
-		timerProcessor:     s.mockTimerProcessor,
+		currentClusterName:    s.mockShard.Resource.GetClusterMetadata().GetCurrentClusterName(),
+		shard:                 s.mockShard,
+		clusterMetadata:       s.mockClusterMetadata,
+		executionManager:      s.mockExecutionMgr,
+		historyCache:          historyCache,
+		logger:                s.logger,
+		tokenSerializer:       common.NewProtoTaskTokenSerializer(),
+		metricsClient:         s.mockShard.GetMetricsClient(),
+		eventNotifier:         events.NewNotifier(s.timeSource, metrics.NewNoopMetricsClient(), func(namespace.ID, string) int32 { return 1 }),
+		txProcessor:           s.mockTxProcessor,
+		timerProcessor:        s.mockTimerProcessor,
+		workflowDeleteManager: s.mockDeleteManager,
 	}
 	s.mockShard.SetEngineForTesting(h)
 	s.mockBean = client.NewMockBean(s.controller)
 	s.mockBean.EXPECT().GetRemoteAdminClient("standby").Return(s.mockAdminClient)
 
 	s.timerQueueStandbyTaskExecutor = newTimerQueueStandbyTaskExecutor(
 		s.mockShard,
-		h,
+		s.mockDeleteManager,
+		historyCache,
 		s.mockNDCHistoryResender,
 		s.logger,
 		s.mockShard.GetMetricsClient(),