Delete old archival code

common/archiver/archivalMetadata.go

@@ -79,9 +79,7 @@ func (a *archivalConfig) StaticClusterState() ArchivalState {
 const (
 	// ArchivalDisabled means this cluster is not configured to handle archival
 	ArchivalDisabled ArchivalState = iota
-	// ArchivalPaused means this cluster is configured to handle archival but is currently not archiving
-	// This state is not yet implemented, as of now ArchivalPaused is treated the same way as ArchivalDisabled
-	ArchivalPaused
+	_                              // for deprecated ArchivalPaused state
 	// ArchivalEnabled means this cluster is currently archiving
 	ArchivalEnabled
 )
@@ -216,8 +214,6 @@ func getClusterArchivalState(str string) (ArchivalState, error) {
 	switch str {
 	case "", config.ArchivalDisabled:
 		return ArchivalDisabled, nil
-	case config.ArchivalPaused:
-		return ArchivalPaused, nil
 	case config.ArchivalEnabled:
 		return ArchivalEnabled, nil
 	}

common/dynamicconfig/constants.go

@@ -538,8 +538,6 @@ const (
 	ArchivalProcessorArchiveDelay = "history.archivalProcessorArchiveDelay"
 	// ArchivalBackendMaxRPS is the maximum rate of requests per second to the archival backend
 	ArchivalBackendMaxRPS = "history.archivalBackendMaxRPS"
-	// DurableArchivalEnabled is the flag to enable durable archival
-	DurableArchivalEnabled = "history.durableArchivalEnabled"
 
 	// ReplicatorTaskBatchSize is batch size for ReplicatorProcessor
 	ReplicatorTaskBatchSize = "history.replicatorTaskBatchSize"

common/persistence/serialization/task_serializer.go

@@ -550,33 +550,22 @@ func (s *TaskSerializer) transferCloseTaskToProto(
 		TaskId:                  closeTask.TaskID,
 		VisibilityTime:          timestamp.TimePtr(closeTask.VisibilityTimestamp),
 		DeleteAfterClose:        closeTask.DeleteAfterClose,
-		TaskDetails: &persistencespb.TransferTaskInfo_CloseExecutionTaskDetails_{
-			CloseExecutionTaskDetails: &persistencespb.TransferTaskInfo_CloseExecutionTaskDetails{
-				CanSkipVisibilityArchival: closeTask.CanSkipVisibilityArchival,
-			},
-		},
 	}
 }
 
 func (s *TaskSerializer) transferCloseTaskFromProto(
 	closeTask *persistencespb.TransferTaskInfo,
 ) *tasks.CloseExecutionTask {
-	canSkipVisibilityArchival := false
-	closeExecutionTaskDetails := closeTask.GetCloseExecutionTaskDetails()
-	if closeExecutionTaskDetails != nil {
-		canSkipVisibilityArchival = closeExecutionTaskDetails.CanSkipVisibilityArchival
-	}
 	return &tasks.CloseExecutionTask{
 		WorkflowKey: definition.NewWorkflowKey(
 			closeTask.NamespaceId,
 			closeTask.WorkflowId,
 			closeTask.RunId,
 		),
-		VisibilityTimestamp:       *closeTask.VisibilityTime,
-		TaskID:                    closeTask.TaskId,
-		Version:                   closeTask.Version,
-		DeleteAfterClose:          closeTask.DeleteAfterClose,
-		CanSkipVisibilityArchival: canSkipVisibilityArchival,
+		VisibilityTimestamp: *closeTask.VisibilityTime,
+		TaskID:              closeTask.TaskId,
+		Version:             closeTask.Version,
+		DeleteAfterClose:    closeTask.DeleteAfterClose,
 		// Delete workflow task process stage is not persisted. It is only for in memory retries.
 		DeleteProcessStage: tasks.DeleteWorkflowExecutionStageNone,
 	}
@@ -872,7 +861,6 @@ func (s *TaskSerializer) timerWorkflowCleanupTaskToProto(
 		TaskId:              workflowCleanupTimer.TaskID,
 		VisibilityTime:      &workflowCleanupTimer.VisibilityTimestamp,
 		BranchToken:         workflowCleanupTimer.BranchToken,
-		AlreadyArchived:     workflowCleanupTimer.WorkflowDataAlreadyArchived,
 	}
 }
 
@@ -885,11 +873,10 @@ func (s *TaskSerializer) timerWorkflowCleanupTaskFromProto(
 			workflowCleanupTimer.WorkflowId,
 			workflowCleanupTimer.RunId,
 		),
-		VisibilityTimestamp:         *workflowCleanupTimer.VisibilityTime,
-		TaskID:                      workflowCleanupTimer.TaskId,
-		Version:                     workflowCleanupTimer.Version,
-		BranchToken:                 workflowCleanupTimer.BranchToken,
-		WorkflowDataAlreadyArchived: workflowCleanupTimer.AlreadyArchived,
+		VisibilityTimestamp: *workflowCleanupTimer.VisibilityTime,
+		TaskID:              workflowCleanupTimer.TaskId,
+		Version:             workflowCleanupTimer.Version,
+		BranchToken:         workflowCleanupTimer.BranchToken,
 		// Delete workflow task process stage is not persisted. It is only for in memory retries.
 		ProcessStage: tasks.DeleteWorkflowExecutionStageNone,
 	}

common/persistence/serialization/task_serializer_test.go

@@ -158,7 +158,6 @@ func (s *taskSerializerSuite) TestTransferCloseTask() {
 	}
 	s.assertEqualTasks(closeTask)
 
-	closeTask.CanSkipVisibilityArchival = true
 	s.assertEqualTasks(closeTask)
 }
 
@@ -258,8 +257,6 @@ func (s *taskSerializerSuite) TestTimerWorkflowCleanupTask() {
 		BranchToken:         []byte{123},
 	}
 	s.assertEqualTasks(workflowCleanupTimer)
-	workflowCleanupTimer.WorkflowDataAlreadyArchived = true
-	s.assertEqualTasks(workflowCleanupTimer)
 }
 
 func (s *taskSerializerSuite) TestVisibilityStartTask() {

service/history/archival_queue_task_executor.go

@@ -251,7 +251,7 @@ func (e *archivalQueueTaskExecutor) addDeletionTask(
 		e.shardContext.GetConfig(),
 		e.shardContext.GetArchivalMetadata(),
 	)
-	err = taskGenerator.GenerateDeleteHistoryEventTask(*closeTime, true)
+	err = taskGenerator.GenerateDeleteHistoryEventTask(*closeTime)
 	if err != nil {
 		return err
 	}

service/history/archival_queue_task_executor_test.go

@@ -433,7 +433,6 @@ func TestArchivalQueueTaskExecutor(t *testing.T) {
 						assert.Zero(t, task.TaskID)
 						assert.Equal(t, p.LastWriteVersionBeforeArchival, task.Version)
 						assert.Equal(t, branchToken, task.BranchToken)
-						assert.True(t, task.WorkflowDataAlreadyArchived)
 						assert.Equal(t, p.ExpectedDeleteTime, task.VisibilityTimestamp)
 						popTasks := map[tasks.Category][]tasks.Task{
 							tasks.CategoryTimer: {

service/history/configs/config.go

@@ -178,7 +178,6 @@ type Config struct {
 	NumArchiveSystemWorkflows dynamicconfig.IntPropertyFn
 	ArchiveRequestRPS         dynamicconfig.IntPropertyFn
 	ArchiveSignalTimeout      dynamicconfig.DurationPropertyFn
-	DurableArchivalEnabled    dynamicconfig.BoolPropertyFn
 
 	// Size limit related settings
 	BlobSizeLimitError               dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -417,7 +416,6 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis
 		NumArchiveSystemWorkflows: dc.GetIntProperty(dynamicconfig.NumArchiveSystemWorkflows, 1000),
 		ArchiveRequestRPS:         dc.GetIntProperty(dynamicconfig.ArchiveRequestRPS, 300), // should be much smaller than frontend RPS
 		ArchiveSignalTimeout:      dc.GetDurationProperty(dynamicconfig.ArchiveSignalTimeout, 300*time.Millisecond),
-		DurableArchivalEnabled:    dc.GetBoolProperty(dynamicconfig.DurableArchivalEnabled, true),
 
 		BlobSizeLimitError:               dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitError, 2*1024*1024),
 		BlobSizeLimitWarn:                dc.GetIntPropertyFilteredByNamespace(dynamicconfig.BlobSizeLimitWarn, 512*1024),

service/history/deletemanager/delete_manager.go

@@ -31,22 +31,18 @@ import (
 	"time"
 
 	commonpb "go.temporal.io/api/common/v1"
-	enumspb "go.temporal.io/api/enums/v1"
 
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
-	"go.temporal.io/server/common/primitives"
-	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/tasks"
 	"go.temporal.io/server/service/history/workflow"
 	wcache "go.temporal.io/server/service/history/workflow/cache"
-	"go.temporal.io/server/service/worker/archiver"
 )
 
 type (
@@ -73,7 +69,6 @@ type (
 			we commonpb.WorkflowExecution,
 			weCtx workflow.Context,
 			ms workflow.MutableState,
-			archiveIfEnabled bool,
 			stage *tasks.DeleteWorkflowExecutionStage,
 		) error
 	}
@@ -83,7 +78,6 @@ type (
 		workflowCache  wcache.Cache
 		config         *configs.Config
 		metricsHandler metrics.Handler
-		archivalClient archiver.Client
 		timeSource     clock.TimeSource
 	}
 )
@@ -94,15 +88,13 @@ func NewDeleteManager(
 	shard shard.Context,
 	cache wcache.Cache,
 	config *configs.Config,
-	archiverClient archiver.Client,
 	timeSource clock.TimeSource,
 ) *DeleteManagerImpl {
 	deleteManager := &DeleteManagerImpl{
 		shard:          shard,
 		workflowCache:  cache,
 		metricsHandler: shard.GetMetricsHandler(),
 		config:         config,
-		archivalClient: archiverClient,
 		timeSource:     timeSource,
 	}
 
@@ -155,7 +147,6 @@ func (m *DeleteManagerImpl) DeleteWorkflowExecution(
 		we,
 		weCtx,
 		ms,
-		false,
 		forceDeleteFromOpenVisibility,
 		stage,
 		m.metricsHandler.WithTags(metrics.OperationTag(metrics.HistoryDeleteWorkflowExecutionScope)),
@@ -168,7 +159,6 @@ func (m *DeleteManagerImpl) DeleteWorkflowExecutionByRetention(
 	we commonpb.WorkflowExecution,
 	weCtx workflow.Context,
 	ms workflow.MutableState,
-	archiveIfEnabled bool,
 	stage *tasks.DeleteWorkflowExecutionStage,
 ) error {
 
@@ -178,7 +168,6 @@ func (m *DeleteManagerImpl) DeleteWorkflowExecutionByRetention(
 		we,
 		weCtx,
 		ms,
-		archiveIfEnabled,
 		false, // When retention is fired, workflow execution is always closed.
 		stage,
 		m.metricsHandler.WithTags(metrics.OperationTag(metrics.HistoryProcessDeleteHistoryEventScope)),
@@ -191,7 +180,6 @@ func (m *DeleteManagerImpl) deleteWorkflowExecutionInternal(
 	we commonpb.WorkflowExecution,
 	weCtx workflow.Context,
 	ms workflow.MutableState,
-	archiveIfEnabled bool,
 	forceDeleteFromOpenVisibility bool,
 	stage *tasks.DeleteWorkflowExecutionStage,
 	metricsHandler metrics.Handler,
@@ -218,26 +206,6 @@ func (m *DeleteManagerImpl) deleteWorkflowExecutionInternal(
 		}
 	}
 
-	// NOTE: old versions (before server version 1.17.3) of archival workflow will delete workflow history directly
-	// after archiving history. But getting workflow close time requires workflow close event (for workflows closed by
-	// server version before 1.17), so this step needs to be done after getting workflow close time.
-	if archiveIfEnabled {
-		deletionPromised, err := m.archiveWorkflowIfEnabled(ctx, namespaceID, we, currentBranchToken, weCtx, ms, metricsHandler)
-		if err != nil {
-			return err
-		}
-		if deletionPromised {
-			// Don't delete workflow data. The workflow data will be deleted after history archived.
-			// if we proceed to delete mutable state, then history scavanger may kick in and
-			// delete history before history archival is done.
-
-			// HOWEVER, when rolling out this change, we don't know if worker is running an old version of the
-			// archival workflow (before 1.17.3), which will only delete workflow history. To prevent this from
-			// happening, worker role must be deployed first.
-			return nil
-		}
-	}
-
 	if err := m.shard.DeleteWorkflowExecution(
 		ctx,
 		definition.WorkflowKey{
@@ -260,75 +228,3 @@ func (m *DeleteManagerImpl) deleteWorkflowExecutionInternal(
 	metricsHandler.Counter(metrics.WorkflowCleanupDeleteCount.GetMetricName()).Record(1)
 	return nil
 }
-
-func (m *DeleteManagerImpl) archiveWorkflowIfEnabled(
-	ctx context.Context,
-	namespaceID namespace.ID,
-	workflowExecution commonpb.WorkflowExecution,
-	currentBranchToken []byte,
-	weCtx workflow.Context,
-	ms workflow.MutableState,
-	metricsHandler metrics.Handler,
-) (deletionPromised bool, err error) {
-
-	namespaceRegistryEntry := ms.GetNamespaceEntry()
-
-	clusterConfiguredForHistoryArchival := m.shard.GetArchivalMetadata().GetHistoryConfig().ClusterConfiguredForArchival()
-	namespaceConfiguredForHistoryArchival := namespaceRegistryEntry.HistoryArchivalState().State == enumspb.ARCHIVAL_STATE_ENABLED
-	archiveHistory := clusterConfiguredForHistoryArchival && namespaceConfiguredForHistoryArchival
-
-	// TODO: @ycyang once archival backfill is in place cluster:paused && namespace:enabled should be a nop rather than a delete
-	if !archiveHistory {
-		return false, nil
-	}
-
-	closeFailoverVersion, err := ms.GetLastWriteVersion()
-	if err != nil {
-		return false, err
-	}
-
-	req := &archiver.ClientRequest{
-		ArchiveRequest: &archiver.ArchiveRequest{
-			ShardID:              m.shard.GetShardID(),
-			NamespaceID:          namespaceID.String(),
-			WorkflowID:           workflowExecution.GetWorkflowId(),
-			RunID:                workflowExecution.GetRunId(),
-			Namespace:            namespaceRegistryEntry.Name().String(),
-			Targets:              []archiver.ArchivalTarget{archiver.ArchiveTargetHistory},
-			HistoryURI:           namespaceRegistryEntry.HistoryArchivalState().URI,
-			NextEventID:          ms.GetNextEventID(),
-			BranchToken:          currentBranchToken,
-			CloseFailoverVersion: closeFailoverVersion,
-		},
-		CallerService:        string(primitives.HistoryService),
-		AttemptArchiveInline: false, // archive in workflow by default
-	}
-	executionStats, err := weCtx.LoadExecutionStats(ctx)
-	if err == nil && executionStats.HistorySize < int64(m.config.TimerProcessorHistoryArchivalSizeLimit()) {
-		req.AttemptArchiveInline = true
-	}
-
-	saTypeMap, err := m.shard.GetSearchAttributesProvider().GetSearchAttributes(m.config.DefaultVisibilityIndexName, false)
-	if err != nil {
-		return false, err
-	}
-	// Setting search attributes types here because archival client needs to stringify them,
-	// and it might not have access to typeMap (i.e. type needs to be embedded).
-	searchattribute.ApplyTypeMap(req.ArchiveRequest.SearchAttributes, saTypeMap)
-
-	ctx, cancel := context.WithTimeout(context.Background(), m.config.TimerProcessorArchivalTimeLimit())
-	defer cancel()
-	resp, err := m.archivalClient.Archive(ctx, req)
-	if err != nil {
-		return false, err
-	}
-	if resp.HistoryArchivedInline {
-		metricsHandler.Counter(metrics.WorkflowCleanupDeleteHistoryInlineCount.GetMetricName()).Record(1)
-	} else {
-		metricsHandler.Counter(metrics.WorkflowCleanupArchiveCount.GetMetricName()).Record(1)
-	}
-
-	// inline archival don't perform deletion
-	// only archival through archival workflow will
-	return !resp.HistoryArchivedInline, nil
-}