Delete workflow execution from visibility store even if it was deleted from main store (#3962)

Author: alexshtin

service/frontend/service.go

@@ -231,7 +231,7 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int32, enableReadF
 		DeleteNamespacePageSize:                             dc.GetIntProperty(dynamicconfig.DeleteNamespacePageSize, 1000),
 		DeleteNamespacePagesPerExecution:                    dc.GetIntProperty(dynamicconfig.DeleteNamespacePagesPerExecution, 256),
 		DeleteNamespaceConcurrentDeleteExecutionsActivities: dc.GetIntProperty(dynamicconfig.DeleteNamespaceConcurrentDeleteExecutionsActivities, 4),
-		DeleteNamespaceNamespaceDeleteDelay:                 dc.GetDurationProperty(dynamicconfig.DeleteNamespaceNamespaceDeleteDelay, 0),
+		DeleteNamespaceNamespaceDeleteDelay:                 dc.GetDurationProperty(dynamicconfig.DeleteNamespaceNamespaceDeleteDelay, 0*time.Hour),
 
 		EnableSchedules: dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.FrontendEnableSchedules, true),
 

service/worker/deletenamespace/deleteexecutions/activities.go

@@ -28,6 +28,7 @@ import (
 	"context"
 
 	"go.temporal.io/api/serviceerror"
+	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/sdk/activity"
 
 	"go.temporal.io/server/api/historyservice/v1"
@@ -137,9 +138,18 @@ func (a *Activities) DeleteExecutionsActivity(ctx context.Context, params Delete
 		case nil:
 			result.SuccessCount++
 			a.metricsHandler.Counter(metrics.DeleteExecutionsSuccessCount.GetMetricName()).Record(1)
-		case *serviceerror.NotFound: // Workflow execution doesn't exist. Do nothing.
+
+		case *serviceerror.NotFound:
 			a.metricsHandler.Counter(metrics.DeleteExecutionNotFoundCount.GetMetricName()).Record(1)
-			a.logger.Info("Workflow execution is not found.", tag.WorkflowNamespace(params.Namespace.String()), tag.WorkflowID(execution.Execution.GetWorkflowId()), tag.WorkflowRunID(execution.Execution.GetRunId()))
+			a.logger.Info("Workflow execution is not found in history service.", tag.WorkflowNamespace(params.Namespace.String()), tag.WorkflowID(execution.Execution.GetWorkflowId()), tag.WorkflowRunID(execution.Execution.GetRunId()))
+			// The reasons why workflow execution doesn't exist in history service, but exists in visibility store might be:
+			// 1. The workflow execution was deleted by someone else after last ListWorkflowExecutions call but before historyClient.DeleteWorkflowExecution call.
+			// 2. Database is in inconsistent state: workflow execution was manually deleted from history store, but not from visibility store.
+			// To avoid continuously getting this workflow execution from visibility store, it needs to be deleted directly from visibility store.
+			s, e := a.deleteWorkflowExecutionFromVisibility(ctx, params.NamespaceID, execution)
+			result.SuccessCount += s
+			result.ErrorCount += e
+
 		default:
 			result.ErrorCount++
 			a.metricsHandler.Counter(metrics.DeleteExecutionFailuresCount.GetMetricName()).Record(1)
@@ -154,3 +164,33 @@ func (a *Activities) DeleteExecutionsActivity(ctx context.Context, params Delete
 	}
 	return result, nil
 }
+
+func (a *Activities) deleteWorkflowExecutionFromVisibility(
+	ctx context.Context,
+	namespaceID namespace.ID,
+	execution *workflowpb.WorkflowExecutionInfo,
+) (successCount int, errorCount int) {
+
+	a.logger.Info("Deleting workflow execution from visibility.", tag.WorkflowNamespaceID(namespaceID.String()), tag.WorkflowID(execution.Execution.GetWorkflowId()), tag.WorkflowRunID(execution.Execution.GetRunId()))
+	_, err := a.historyClient.DeleteWorkflowVisibilityRecord(ctx, &historyservice.DeleteWorkflowVisibilityRecordRequest{
+		NamespaceId:       namespaceID.String(),
+		Execution:         execution.GetExecution(),
+		WorkflowStartTime: execution.GetStartTime(),
+		WorkflowCloseTime: execution.GetCloseTime(),
+	})
+	switch err.(type) {
+	case nil:
+		// Indicates that main and visibility stores were in inconsistent state.
+		a.metricsHandler.Counter(metrics.DeleteExecutionsSuccessCount.GetMetricName()).Record(1)
+		a.logger.Info("Workflow execution deleted from visibility.", tag.WorkflowNamespaceID(namespaceID.String()), tag.WorkflowID(execution.Execution.GetWorkflowId()), tag.WorkflowRunID(execution.Execution.GetRunId()))
+		return 1, 0
+	case *serviceerror.NotFound:
+		// Indicates that workflow execution was deleted by someone else.
+		a.logger.Error("Workflow execution is not found in visibility store.", tag.WorkflowNamespaceID(namespaceID.String()), tag.WorkflowID(execution.Execution.GetWorkflowId()), tag.WorkflowRunID(execution.Execution.GetRunId()))
+		return 0, 0
+	default:
+		a.metricsHandler.Counter(metrics.DeleteExecutionFailuresCount.GetMetricName()).Record(1)
+		a.logger.Error("Unable to delete workflow execution from visibility store.", tag.WorkflowNamespaceID(namespaceID.String()), tag.WorkflowID(execution.Execution.GetWorkflowId()), tag.WorkflowRunID(execution.Execution.GetRunId()), tag.Error(err))
+		return 0, 1
+	}
+}

service/worker/deletenamespace/deleteexecutions/workflow.go

@@ -172,6 +172,7 @@ func DeleteExecutionsWorkflow(ctx workflow.Context, params DeleteExecutionsParam
 		}
 	}
 
+	// If nextPageToken is nil then there are no more workflow executions to delete.
 	if nextPageToken == nil {
 		if result.ErrorCount == 0 {
 			logger.Info("Successfully deleted workflow executions.", tag.WorkflowNamespace(params.Namespace.String()), tag.DeletedExecutionsCount(result.SuccessCount))
@@ -181,7 +182,7 @@ func DeleteExecutionsWorkflow(ctx workflow.Context, params DeleteExecutionsParam
 		return result, nil
 	}
 
-	// Too many workflow executions, and ConcurrentDeleteExecutionsActivities activities has been started already.
+	// Too many workflow executions, and ConcurrentDeleteExecutionsActivities number of activities has been completed already.
 	// Continue as new to prevent workflow history size explosion.
 
 	params.PreviousSuccessCount = result.SuccessCount

service/worker/deletenamespace/deleteexecutions/workflow_test.go

@@ -190,7 +190,7 @@ func Test_DeleteExecutionsWorkflow_ManyExecutions_ContinueAsNew(t *testing.T) {
 	var a *Activities
 
 	env.OnActivity(a.GetNextPageTokenActivity, mock.Anything, mock.Anything).Return([]byte{3, 22, 83}, nil).Times(78)
-	env.OnActivity(a.DeleteExecutionsActivity, mock.Anything, mock.Anything).Return(DeleteExecutionsActivityResult{SuccessCount: 1}, nil).Times(78)
+	env.OnActivity(a.DeleteExecutionsActivity, mock.Anything, mock.Anything).Return(DeleteExecutionsActivityResult{SuccessCount: 1, ErrorCount: 0}, nil).Times(78)
 
 	env.ExecuteWorkflow(DeleteExecutionsWorkflow, DeleteExecutionsParams{
 		NamespaceID: "namespace-id",
@@ -306,8 +306,12 @@ func Test_DeleteExecutionsWorkflow_NoActivityMocks_ManyExecutions(t *testing.T)
 
 	historyClient := historyservicemock.NewMockHistoryServiceClient(ctrl)
 	historyClient.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, nil).Times(2)
-	// NotFound errors should not affect the error count.
+
 	historyClient.EXPECT().DeleteWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewNotFound("not found")).Times(2)
+	// NotFound errors should not affect neither error nor success count.
+	historyClient.EXPECT().DeleteWorkflowVisibilityRecord(gomock.Any(), gomock.Any()).Return(nil, serviceerror.NewNotFound("not found"))
+	// NotFound in main store but no error from visibility store adds 1 to success count.
+	historyClient.EXPECT().DeleteWorkflowVisibilityRecord(gomock.Any(), gomock.Any()).Return(nil, nil)
 
 	a := &Activities{
 		visibilityManager: visibilityManager,
@@ -333,7 +337,7 @@ func Test_DeleteExecutionsWorkflow_NoActivityMocks_ManyExecutions(t *testing.T)
 	var result DeleteExecutionsResult
 	require.NoError(t, env.GetWorkflowResult(&result))
 	require.Equal(t, 0, result.ErrorCount)
-	require.Equal(t, 2, result.SuccessCount)
+	require.Equal(t, 3, result.SuccessCount)
 }
 
 func Test_DeleteExecutionsWorkflow_NoActivityMocks_HistoryClientError(t *testing.T) {

tests/namespace_delete_test.go

@@ -43,9 +43,11 @@ import (
 	"go.temporal.io/api/workflowservice/v1"
 
 	"go.temporal.io/server/api/adminservice/v1"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/sql/sqlplugin/mysql"
 	"go.temporal.io/server/common/persistence/sql/sqlplugin/postgresql"
 	"go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite"
@@ -190,24 +192,28 @@ func (s *namespaceTestSuite) Test_NamespaceDelete_WithWorkflows() {
 	nsID := descResp.GetNamespaceInfo().GetId()
 
 	// Start few workflow executions.
+	var executions []*commonpb.WorkflowExecution
 	for i := 0; i < 100; i++ {
-		_, err = s.frontendClient.StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		wid := "wf_id_" + strconv.Itoa(i)
+		resp, err := s.frontendClient.StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
 			RequestId:    uuid.New(),
 			Namespace:    "ns_name_los_angeles",
-			WorkflowId:   "wf_id_" + strconv.Itoa(i),
+			WorkflowId:   wid,
 			WorkflowType: &commonpb.WorkflowType{Name: "workflowTypeName"},
 			TaskQueue:    &taskqueuepb.TaskQueue{Name: "taskQueueName"},
 		})
 		s.NoError(err)
+		executions = append(executions, &commonpb.WorkflowExecution{
+			WorkflowId: wid,
+			RunId:      resp.GetRunId(),
+		})
 	}
 
 	// Terminate some workflow executions.
-	for i := 0; i < 30; i++ {
+	for _, execution := range executions[:30] {
 		_, err = s.frontendClient.TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
-			Namespace: "ns_name_los_angeles",
-			WorkflowExecution: &commonpb.WorkflowExecution{
-				WorkflowId: "wf_id_" + strconv.Itoa(i),
-			},
+			Namespace:         "ns_name_los_angeles",
+			WorkflowExecution: execution,
 		})
 		s.NoError(err)
 	}
@@ -224,37 +230,123 @@ func (s *namespaceTestSuite) Test_NamespaceDelete_WithWorkflows() {
 	s.NoError(err)
 	s.Equal(enumspb.NAMESPACE_STATE_DELETED, descResp2.GetNamespaceInfo().GetState())
 
-	namespaceExistsOp := func() error {
+	s.Eventually(func() bool {
 		_, err := s.frontendClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
 			Id: nsID,
 		})
 		var notFound *serviceerror.NamespaceNotFound
-		if errors.As(err, &notFound) {
-			_, err0 := s.frontendClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
-				Namespace: "ns_name_los_angeles",
-				Execution: &commonpb.WorkflowExecution{
-					WorkflowId: "wf_id_0",
-				},
-			})
-			_, err99 := s.frontendClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+		if !errors.As(err, &notFound) {
+			return false // namespace still exists
+		}
+
+		for _, execution := range executions {
+			_, err = s.frontendClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
 				Namespace: "ns_name_los_angeles",
 				Execution: &commonpb.WorkflowExecution{
-					WorkflowId: "wf_id_99",
+					WorkflowId: execution.GetWorkflowId(),
 				},
 			})
-			if errors.As(err0, &notFound) && errors.As(err99, &notFound) {
-				return nil
+			if !errors.As(err, &notFound) {
+				return false // should never happen
 			}
 		}
-		return errors.New("namespace still exists")
+		return true
+	}, 20*time.Second, time.Second)
+}
+
+func (s *namespaceTestSuite) Test_NamespaceDelete_WithMissingWorkflows() {
+	ctx, cancel := rpc.NewContextWithTimeoutAndVersionHeaders(10000 * time.Second)
+	defer cancel()
+
+	retention := 24 * time.Hour
+	_, err := s.frontendClient.RegisterNamespace(ctx, &workflowservice.RegisterNamespaceRequest{
+		Namespace:                        "ns_name_los_angeles",
+		Description:                      "Namespace to delete",
+		WorkflowExecutionRetentionPeriod: &retention,
+		HistoryArchivalState:             enumspb.ARCHIVAL_STATE_DISABLED,
+		VisibilityArchivalState:          enumspb.ARCHIVAL_STATE_DISABLED,
+	})
+	s.NoError(err)
+	// DescribeNamespace reads directly from database but namespace validator uses cache.
+	s.cluster.RefreshNamespaceCache()
+
+	descResp, err := s.frontendClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+		Namespace: "ns_name_los_angeles",
+	})
+	s.NoError(err)
+	nsID := descResp.GetNamespaceInfo().GetId()
+
+	// Start few workflow executions.
+
+	var executions []*commonpb.WorkflowExecution
+	for i := 0; i < 10; i++ {
+		wid := "wf_id_" + strconv.Itoa(i)
+		resp, err := s.frontendClient.StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+			RequestId:    uuid.New(),
+			Namespace:    "ns_name_los_angeles",
+			WorkflowId:   wid,
+			WorkflowType: &commonpb.WorkflowType{Name: "workflowTypeName"},
+			TaskQueue:    &taskqueuepb.TaskQueue{Name: "taskQueueName"},
+		})
+		s.NoError(err)
+		executions = append(executions, &commonpb.WorkflowExecution{
+			WorkflowId: wid,
+			RunId:      resp.GetRunId(),
+		})
 	}
 
-	namespaceExistsPolicy := backoff.NewExponentialRetryPolicy(time.Second).
-		WithBackoffCoefficient(1).
-		WithExpirationInterval(30 * time.Second)
+	// Delete some workflow executions from DB but not from visibility.
+	// Every subsequent delete (from deleteexecutions.Workflow) from ES will take at least 1s due to bulk processor.
+	for _, execution := range executions[0:5] {
+		shardID := common.WorkflowIDToHistoryShard(
+			nsID,
+			execution.GetWorkflowId(),
+			s.clusterConfig.HistoryConfig.NumHistoryShards,
+		)
+
+		err = s.cluster.GetExecutionManager().DeleteWorkflowExecution(ctx, &persistence.DeleteWorkflowExecutionRequest{
+			ShardID:     shardID,
+			NamespaceID: nsID,
+			WorkflowID:  execution.GetWorkflowId(),
+			RunID:       execution.GetRunId(),
+		})
+		s.NoError(err)
+	}
 
-	err = backoff.ThrottleRetry(namespaceExistsOp, namespaceExistsPolicy, func(_ error) bool { return true })
+	delResp, err := s.operatorClient.DeleteNamespace(ctx, &operatorservice.DeleteNamespaceRequest{
+		Namespace: "ns_name_los_angeles",
+	})
+	s.NoError(err)
+	s.Equal("ns_name_los_angeles-deleted-"+nsID[:5], delResp.GetDeletedNamespace())
+
+	descResp2, err := s.frontendClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+		Id: nsID,
+	})
 	s.NoError(err)
+	s.Equal(enumspb.NAMESPACE_STATE_DELETED, descResp2.GetNamespaceInfo().GetState())
+
+	s.Eventually(func() bool {
+		_, err := s.frontendClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
+			Id: nsID,
+		})
+		var notFound *serviceerror.NamespaceNotFound
+		if !errors.As(err, &notFound) {
+			return false // namespace still exists
+		}
+
+		for _, execution := range executions {
+			_, err = s.frontendClient.DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
+				Namespace: "ns_name_los_angeles",
+				Execution: &commonpb.WorkflowExecution{
+					WorkflowId: execution.GetWorkflowId(),
+				},
+			})
+			if !errors.As(err, &notFound) {
+				return false // should never happen
+			}
+		}
+		return true
+	}, 20*time.Second, time.Second)
 }
 
 func (s *namespaceTestSuite) Test_NamespaceDelete_CrossNamespaceChild() {