Fix history scavenger bug on delete mutable state (#3588)

* Check workflow state in scavenger * Log deleted workflow as record
temporalio · Nov 15, 2022 · e7f99d0 · e7f99d0
1 parent 4f2b98e
commit e7f99d0
Show file tree

Hide file tree

Showing 3 changed files with 42 additions and 2 deletions.
diff --git a/service/worker/scanner/executions/task.go b/service/worker/scanner/executions/task.go
@@ -263,7 +263,7 @@ func printValidationResult(
 	handler.Counter(metrics.ScavengerValidationFailuresCount.GetMetricName()).Record(1)
 	for _, result := range results {
 		handler.Counter(metrics.ScavengerValidationFailuresCount.GetMetricName()).Record(1, metrics.FailureTag(result.failureType))
-		logger.Error(
+		logger.Info(
 			"validation failed for execution.",
 			tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
 			tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),

diff --git a/service/worker/scanner/history/scavenger.go b/service/worker/scanner/history/scavenger.go
@@ -34,6 +34,7 @@ import (
 	"go.temporal.io/sdk/activity"
 
 	"go.temporal.io/server/api/adminservice/v1"
+	"go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencepb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common"
@@ -339,6 +340,11 @@ func (s *Scavenger) cleanUpWorkflowPastRetention(
 	ctx context.Context,
 	mutableState *persistencepb.WorkflowMutableState,
 ) error {
+	if mutableState.GetExecutionState().GetState() != enums.WORKFLOW_EXECUTION_STATE_COMPLETED {
+		// Skip running workflow
+		return nil
+	}
+
 	executionInfo := mutableState.GetExecutionInfo()
 	ns, err := s.registry.GetNamespaceByID(namespace.ID(executionInfo.GetNamespaceId()))
 	switch err.(type) {
@@ -370,7 +376,13 @@ func (s *Scavenger) cleanUpWorkflowPastRetention(
 				tag.WorkflowID(executionInfo.GetWorkflowId()),
 				tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
 			)
+			return nil
 		}
+		s.logger.Info("Delete workflow data past retention via history scavenger",
+			tag.WorkflowNamespace(ns.Name().String()),
+			tag.WorkflowID(executionInfo.GetWorkflowId()),
+			tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
+		)
 	}
 	return nil
 }

diff --git a/service/worker/scanner/history/scavenger_test.go b/service/worker/scanner/history/scavenger_test.go
@@ -38,6 +38,7 @@ import (
 
 	"go.temporal.io/server/api/adminservice/v1"
 	"go.temporal.io/server/api/adminservicemock/v1"
+	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/historyservicemock/v1"
 	persistencepb "go.temporal.io/server/api/persistence/v1"
@@ -554,6 +555,11 @@ func (s *ScavengerTestSuite) TestDeleteWorkflowAfterRetention() {
 				ForkTime:    timestamp.TimeNowPtrUtcAddDuration(-retention * 2),
 				Info:        persistence.BuildHistoryGarbageCleanupInfo("namespaceID4", "workflowID4", "runID4"),
 			},
+			{
+				BranchToken: s.toBranchToken("treeID5", "branchID5"),
+				ForkTime:    timestamp.TimeNowPtrUtcAddDuration(-retention * 2),
+				Info:        persistence.BuildHistoryGarbageCleanupInfo("namespaceID5", "workflowID5", "runID5"),
+			},
 		},
 	}, nil)
 	mockedNamespace := namespace.NewNamespaceForTest(
@@ -579,6 +585,7 @@ func (s *ScavengerTestSuite) TestDeleteWorkflowAfterRetention() {
 			},
 			ExecutionState: &persistencepb.WorkflowExecutionState{
 				RunId: "runID2",
+				State: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
 			},
 		},
 	}
@@ -591,6 +598,20 @@ func (s *ScavengerTestSuite) TestDeleteWorkflowAfterRetention() {
 			},
 			ExecutionState: &persistencepb.WorkflowExecutionState{
 				RunId: "runID4",
+				State: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
+			},
+		},
+	}
+	runningWorkflow5 := &historyservice.DescribeMutableStateResponse{
+		DatabaseMutableState: &persistencepb.WorkflowMutableState{
+			ExecutionInfo: &persistencepb.WorkflowExecutionInfo{
+				WorkflowId:     "workflowID5",
+				NamespaceId:    "namespaceID5",
+				LastUpdateTime: timestamp.TimePtr(time.Now().UTC().Add(-time.Hour * 24)),
+			},
+			ExecutionState: &persistencepb.WorkflowExecutionState{
+				RunId: "runID5",
+				State: enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
 			},
 		},
 	}
@@ -623,6 +644,13 @@ func (s *ScavengerTestSuite) TestDeleteWorkflowAfterRetention() {
 			RunId:      "runID4",
 		},
 	}).Return(workflowPastRetention4, nil)
+	s.mockHistoryClient.EXPECT().DescribeMutableState(gomock.Any(), &historyservice.DescribeMutableStateRequest{
+		NamespaceId: "namespaceID5",
+		Execution: &commonpb.WorkflowExecution{
+			WorkflowId: "workflowID5",
+			RunId:      "runID5",
+		},
+	}).Return(runningWorkflow5, nil)
 	s.mockAdminClient.EXPECT().DeleteWorkflowExecution(gomock.Any(), &adminservice.DeleteWorkflowExecutionRequest{
 		Execution: &commonpb.WorkflowExecution{
 			WorkflowId: "workflowID2",
@@ -639,7 +667,7 @@ func (s *ScavengerTestSuite) TestDeleteWorkflowAfterRetention() {
 	hbd, err := s.scavenger.Run(context.Background())
 	s.Nil(err)
 	s.Equal(0, hbd.SkipCount)
-	s.Equal(4, hbd.SuccessCount)
+	s.Equal(5, hbd.SuccessCount)
 	s.Equal(0, hbd.ErrorCount)
 	s.Equal(2, hbd.CurrentPage)
 	s.Equal(0, len(hbd.NextPageToken))